diff --git a/.env.example b/.env.example index 8ebb4af..c0b3a9a 100644 --- a/.env.example +++ b/.env.example @@ -17,6 +17,10 @@ PROVIDER_URL_KATANA_1=https://rpc.katanarpc.com # Yearn large TVL env vars ENVIO_GRAPHQL_URL="" +# Envio indexer freshness check (protocols/yearn/check_indexer_freshness.py) +# INDEXER_MAX_LAG_MINUTES=60 # alert when a chain's newest indexed block is older than this +# INDEXER_ALERT_COOLDOWN_HOURS=6 # minimum gap between repeat alerts for the same chain + # Telegram API credentials TELEGRAM_BOT_TOKEN_DEFAULT=your-default-bot-token TELEGRAM_CHAT_ID_DEFAULT=your-default-chat-id diff --git a/automation/jobs.yaml b/automation/jobs.yaml index 710b223..da74d10 100644 --- a/automation/jobs.yaml +++ b/automation/jobs.yaml @@ -23,6 +23,10 @@ profiles: # No env override: alert dedupe and morpho rows share the default cache-id.txt # under $CACHE_DIR. tasks: + # Runs first: every Envio-backed monitor below (large flows, timelock, + # 3jane borrower watch) goes quiet rather than loud when the indexer + # stalls, so the freshness check tells us the silence is not good news. + - { name: "yearn-check-indexer-freshness", script: protocols/yearn/check_indexer_freshness.py } - { name: "apyusd", script: protocols/apyusd/main.py } - { name: "3jane", script: protocols/3jane/main.py } - { name: "morpho-markets", script: protocols/morpho/markets.py } diff --git a/monitoring.yaml b/monitoring.yaml index 7a26fb6..1523f90 100644 --- a/monitoring.yaml +++ b/monitoring.yaml @@ -456,7 +456,10 @@ protocols: tasks: - protocols/yearn/alert_large_flows.py - protocols/yearn/check_timelock_delay.py + - 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" - name: "Large Flows" description: "Deposit/withdrawal flows >=$1M USD (or 10% of vault totalSupply fallback for unpriced tokens)" - name: "Timelock Delay" diff --git a/protocols/3jane/main.py b/protocols/3jane/main.py index 3bcd938..691ce06 100644 --- a/protocols/3jane/main.py +++ b/protocols/3jane/main.py @@ -36,7 +36,7 @@ from utils.alert import Alert, AlertSeverity, send_alert from utils.cache import cache_path, get_last_value_for_key_from_file, write_last_value_to_file from utils.chains import Chain -from utils.formatting import format_usd +from utils.formatting import format_duration, format_usd from utils.logger import get_logger from utils.telegram import escape_markdown from utils.web3_wrapper import ChainManager @@ -425,22 +425,6 @@ def format_utc_timestamp(timestamp: int) -> str: return datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") -def format_duration(seconds: int) -> str: - if seconds <= 0: - return "now" - days = seconds // SECONDS_PER_DAY - hours = (seconds % SECONDS_PER_DAY) // 3600 - minutes = (seconds % 3600) // 60 - parts: list[str] = [] - if days: - parts.append(f"{days}d") - if hours: - parts.append(f"{hours}h") - if minutes and not days: - parts.append(f"{minutes}m") - return " ".join(parts) if parts else f"{seconds}s" - - def _borrower_default_cache_key(snapshot: BorrowerRepaymentSnapshot, bucket: str) -> str: return ( f"{CACHE_KEY_BORROWER_DEFAULT_WATCH_PREFIX}:" diff --git a/protocols/yearn/README.md b/protocols/yearn/README.md index 3616221..d26a4f2 100644 --- a/protocols/yearn/README.md +++ b/protocols/yearn/README.md @@ -239,3 +239,44 @@ All chains use the same contract address: `0x88ba032be87d5ef1fbe87336b7090767f36 | Polygon | [polygonscan.com](https://polygonscan.com/address/0x88ba032be87d5ef1fbe87336b7090767f367bf73) | | Katana | [katanascan.com](https://katanascan.com/address/0x88ba032be87d5ef1fbe87336b7090767f367bf73) | | Optimism | [optimistic.etherscan.io](https://optimistic.etherscan.io/address/0x88ba032be87d5ef1fbe87336b7090767f367bf73) | + +======= + +## Indexer Freshness + +The script `yearn/check_indexer_freshness.py` watches the [Envio indexer](https://github.com/chain-events/yearn-indexing-test) that feeds the large-flows, timelock and 3jane borrower monitors. It runs hourly, first in the [hourly profile](../../automation/jobs.yaml). + +An indexer stall is invisible to the monitors that depend on it: GraphQL keeps answering, it just stops returning new rows, so an outage looks exactly like a quiet hour. This check makes the silence loud. + +### How It Works + +1. Queries `chain_metadata` at `ENVIO_GRAPHQL_URL` for each chain's `latest_processed_block`. +2. Fetches that block's timestamp via `ChainManager` and compares it to wall-clock time. +3. Alerts when a chain's newest indexed block is older than `--max-lag-minutes` (default `60`), or when an expected chain reports no sync state at all. + +Step 2 is what makes the check trustworthy. Envio parks `chain_metadata.block_height` at the last processed block once a chain looks caught up, so a stalled indexer keeps reporting itself as zero blocks behind — the same trap called out in the indexer's own [monitoring dashboard](https://envio-monitoring.yearn.dev/). + +Step 3 covers the inverse trap: an empty result set is not good news. If a chain drops out of the indexer's config, or comes back from a restart with no processed block, it simply stops appearing in `chain_metadata` — and a check that only looks at what it was given would report every remaining chain fresh while that chain's monitors sit blind. `EXPECTED_CHAINS` is therefore the authority on what must be present, and anything absent from it alerts. + +`EXPECTED_CHAINS` lists the chains whose indexed events feed monitors here (Mainnet, Optimism, Polygon, Base, Arbitrum, Katana). It is deliberately spelled out rather than derived from the `Chain` enum, so adding an enum member for an unrelated protocol doesn't start alerting that the indexer is missing a chain it was never asked to index — **add a chain here when its events start feeding a monitor.** The indexer also covers Gnosis and Berachain, which nothing here reads from; those are logged and skipped. A chain whose RPC is unreachable is skipped too rather than alerted on: a broken provider is not a stale indexer. + +### Alerts + +All alerts go to the errors channel (`TELEGRAM_*_ERRORS`) labelled `[yearn]`, alongside the other operational diagnostics: + +- **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. +- **Recovered** — sent once when a previously alerting chain catches up. + +A re-sync can run for days, so each chain alerts on the way into trouble and then at most once per `--alert-cooldown-hours` (default `6`) instead of every hourly run. The cooldown is tracked per chain, so one lagging chain never suppresses another's first alert. The last-alert timestamp is cached under `YEARN_INDEXER_STALE_ALERT_`. + +### Usage + +```bash +uv run protocols/yearn/check_indexer_freshness.py +``` + +Optional flags (each also settable via env): + +- `--max-lag-minutes` (default `60`, env `INDEXER_MAX_LAG_MINUTES`) +- `--alert-cooldown-hours` (default `6`, env `INDEXER_ALERT_COOLDOWN_HOURS`) diff --git a/protocols/yearn/check_indexer_freshness.py b/protocols/yearn/check_indexer_freshness.py new file mode 100644 index 0000000..9d3c3c6 --- /dev/null +++ b/protocols/yearn/check_indexer_freshness.py @@ -0,0 +1,359 @@ +#!/usr/bin/env python3 +"""Alert when the Envio indexer stops keeping up with the chain head. + +Several monitors read their events from the Envio indexer (`ENVIO_GRAPHQL_URL`): +Yearn large flows, the timelock alerts and the 3jane borrower watch. When the +indexer stalls they degrade silently — GraphQL keeps answering, it just stops +returning new rows — so an outage is indistinguishable from "nothing happened". + +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 +the indexer reports no sync state for at all, since "nothing was stale" must +never be mistaken for "everything is fresh". + +The RPC round-trip is what makes the check meaningful: envio parks +`chain_metadata.block_height` at the last processed block once a chain looks +caught up, so a stalled indexer keeps reporting itself as zero blocks behind. +The same trap is documented in the indexer's own dashboard +(https://envio-monitoring.yearn.dev/). + +Usage: + python protocols/yearn/check_indexer_freshness.py [--max-lag-minutes 60] +""" + +import argparse +import os +import time +from dataclasses import dataclass + +import requests +from dotenv import load_dotenv + +from utils.cache import cache_filename, get_last_value_for_key_from_file, write_last_value_to_file +from utils.chains import Chain +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.web3_wrapper import ChainManager + +load_dotenv() + +logger = get_logger("yearn.check_indexer_freshness") + +PROTOCOL = "yearn" + +ENVIO_GRAPHQL_URL = os.getenv("ENVIO_GRAPHQL_URL") +DASHBOARD_URL = "https://envio-monitoring.yearn.dev/" + +# A chain is stale once its newest indexed block is older than this. One hour is +# generous for every indexed chain: the slowest of them (Mainnet) produces a +# block every ~12s, so an hour of lag is always a real stall, never jitter. +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. +# 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 + +# Keyed per chain so one lagging chain can't suppress an alert for another. +CACHE_KEY_LAST_ALERT_PREFIX = "YEARN_INDEXER_STALE_ALERT_" + +# Chains whose indexed events feed monitors here. Every one of them must show up +# in chain_metadata — a chain that silently drops out of the indexer's config +# leaves its monitors blind while the endpoint keeps answering, which is exactly +# the failure this script exists to catch. +# +# Kept explicit rather than derived from `Chain` so that adding an enum member +# for an unrelated protocol doesn't start alerting that the indexer is missing a +# chain it was never asked to index. Add a chain here when its events start +# feeding a monitor. +EXPECTED_CHAINS: tuple[Chain, ...] = ( + Chain.MAINNET, + Chain.OPTIMISM, + Chain.POLYGON, + Chain.BASE, + Chain.ARBITRUM, + Chain.KATANA, +) +_EXPECTED_BY_CHAIN_ID: dict[int, Chain] = {chain.chain_id: chain for chain in EXPECTED_CHAINS} + +CHAIN_METADATA_QUERY = """ +{ + chain_metadata { + chain_id + latest_processed_block + block_height + } +} +""" + + +class IndexerUnavailableError(Exception): + """The indexer's GraphQL endpoint could not be queried or returned no chains.""" + + +@dataclass(frozen=True) +class ChainFreshness: + """Freshness of a single indexed chain.""" + + chain: Chain + latest_processed_block: int + # None when the block timestamp could not be resolved, i.e. lag is unknown. + lag_seconds: int | None + + @property + def name(self) -> str: + """Human-readable chain name, e.g. "Mainnet".""" + return self.chain.name.capitalize() + + def is_stale(self, max_lag_seconds: int) -> bool: + """Return True when the newest indexed block is older than the threshold.""" + return self.lag_seconds is not None and self.lag_seconds > max_lag_seconds + + +def fetch_chain_metadata() -> list[dict]: + """Fetch per-chain sync state from the indexer. + + Returns: + The `chain_metadata` rows, one per indexed chain. + + Raises: + IndexerUnavailableError: The endpoint is unset, unreachable, returned + GraphQL errors, or reported no chains at all. + """ + if not ENVIO_GRAPHQL_URL: + raise IndexerUnavailableError( + "ENVIO_GRAPHQL_URL is not set. Set it to the Envio GraphQL endpoint, " + "e.g. export ENVIO_GRAPHQL_URL='https://envio-gql.yearn.dev/v1/graphql'." + ) + + try: + response = request_with_retry("post", ENVIO_GRAPHQL_URL, json={"query": CHAIN_METADATA_QUERY}) + payload = response.json() + except (requests.RequestException, ValueError) as exc: + raise IndexerUnavailableError(f"GraphQL request to {ENVIO_GRAPHQL_URL} failed: {exc}") from exc + + if payload.get("errors"): + raise IndexerUnavailableError(f"GraphQL errors from {ENVIO_GRAPHQL_URL}: {payload['errors']}") + + rows = (payload.get("data") or {}).get("chain_metadata") or [] + if not rows: + raise IndexerUnavailableError(f"chain_metadata is empty at {ENVIO_GRAPHQL_URL} — indexer has no sync state") + return rows + + +def fetch_block_timestamp(chain: Chain, block_number: int) -> int | None: + """Fetch the unix timestamp of a block. + + Args: + chain: Chain the block belongs to. + block_number: Block to look up. + + Returns: + The block timestamp in seconds, or None when the lookup failed — one + unreachable RPC must not mask the other chains' results. + + Note: + Uses the raw JSON-RPC call rather than `eth.get_block`, which rejects the + 97-byte PoA `extraData` on Polygon and pre-Bedrock Optimism blocks — and + old blocks are exactly what a lagging indexer points at. + """ + try: + client = ChainManager.get_client(chain) + response = client.make_request("eth_getBlockByNumber", [hex(block_number), False]) + return int(response["result"]["timestamp"], 16) + except Exception as exc: # noqa: BLE001 - a dead provider is not a stale indexer + logger.warning("Failed to fetch block %d timestamp on %s: %s", block_number, chain.name, exc) + return None + + +def collect_freshness(rows: list[dict], now: int) -> list[ChainFreshness]: + """Resolve how far behind wall-clock time each expected chain is. + + The indexer covers chains this repo doesn't read from (Gnosis, Berachain). + Nothing here consumes their events, so they are skipped rather than alerted + on. Expected chains absent from `rows` produce no entry — see + `missing_chains`. + + Args: + rows: `chain_metadata` rows from the indexer. + now: Current unix timestamp. + + Returns: + One ChainFreshness per expected chain present in `rows`, sorted by chain id. + """ + freshness: list[ChainFreshness] = [] + for row in rows: + chain_id = int(row["chain_id"]) + chain = _EXPECTED_BY_CHAIN_ID.get(chain_id) + if chain is None: + logger.info("Chain %d is indexed but not monitored here, skipping", chain_id) + continue + latest_block = int(row.get("latest_processed_block") or 0) + if latest_block <= 0: + # No processed block means no usable sync state — left out here so + # `missing_chains` reports it rather than treating it as fresh. + logger.warning("%s has no processed block yet", chain.name) + continue + block_timestamp = fetch_block_timestamp(chain, latest_block) + lag = max(0, now - block_timestamp) if block_timestamp is not None else None + lag_text = format_duration(lag) if lag is not None else "unknown" + logger.info("%s: block %d, lag %s", chain.name, latest_block, lag_text) + freshness.append(ChainFreshness(chain=chain, latest_processed_block=latest_block, lag_seconds=lag)) + return sorted(freshness, key=lambda f: f.chain.chain_id) + + +def missing_chains(freshness: list[ChainFreshness]) -> list[Chain]: + """Return expected chains the indexer reported no usable sync state for. + + Covers both a chain absent from `chain_metadata` entirely (dropped from the + indexer config, partial restart) and one present with no processed block. + Either way its monitors are blind, so "nothing was stale" must not be read + as "everything is fresh". + """ + present = {entry.chain for entry in freshness} + return [chain for chain in EXPECTED_CHAINS if chain not in present] + + +def build_alert_message(stale: list[ChainFreshness], missing: list[Chain], max_lag_seconds: int) -> str: + """Build the plain-text alert body for lagging and missing chains.""" + affected = len(stale) + len(missing) + lines = [ + f"Envio indexer problem on {affected} chain(s) — events may be missing from monitoring alerts.", + "", + ] + for entry in stale: + lag = format_duration(entry.lag_seconds or 0) + lines.append( + f"- {entry.name} (chain {entry.chain.chain_id}): {lag} behind, last block {entry.latest_processed_block}" + ) + for chain in missing: + lines.append(f"- {chain.name.capitalize()} (chain {chain.chain_id}): no sync state reported by the indexer") + lines += [ + "", + f"Threshold: {format_duration(max_lag_seconds)}", + f"Dashboard: {DASHBOARD_URL}", + ] + return "\n".join(lines) + + +def _last_alert_timestamp(chain_id: int) -> int: + """Return when this chain last alerted, or 0 if it is currently considered healthy.""" + return int(get_last_value_for_key_from_file(cache_filename, f"{CACHE_KEY_LAST_ALERT_PREFIX}{chain_id}")) + + +def _set_last_alert_timestamp(chain_id: int, timestamp: int) -> None: + """Record the last alert time for a chain (0 clears it back to healthy).""" + write_last_value_to_file(cache_filename, f"{CACHE_KEY_LAST_ALERT_PREFIX}{chain_id}", timestamp) + + +def _is_due_for_alert(chain: Chain, now: int, cooldown_seconds: int) -> bool: + """Return True when this chain has not alerted within the cooldown window.""" + return now - _last_alert_timestamp(chain.chain_id) >= cooldown_seconds + + +def chains_to_alert( + stale: list[ChainFreshness], missing: list[Chain], now: int, cooldown_seconds: int +) -> tuple[list[ChainFreshness], list[Chain]]: + """Filter unhealthy chains down to those outside their re-alert cooldown. + + Args: + stale: Chains currently past the lag threshold. + missing: Expected chains with no usable sync state. + now: Current unix timestamp. + cooldown_seconds: Minimum gap between two alerts for the same chain. + + Returns: + The stale and missing chains that should alert on this run. + """ + return ( + [entry for entry in stale if _is_due_for_alert(entry.chain, now, cooldown_seconds)], + [chain for chain in missing if _is_due_for_alert(chain, now, cooldown_seconds)], + ) + + +def report_recovered(fresh: list[ChainFreshness]) -> None: + """Send a recovery note for chains that had alerted and are now caught up.""" + recovered = [chain for chain in fresh if _last_alert_timestamp(chain.chain.chain_id) > 0] + 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") + for chain in recovered: + _set_last_alert_timestamp(chain.chain.chain_id, 0) + + +def main() -> None: + """Alert on every expected chain the indexer is lagging on or has lost.""" + args = parse_args() + max_lag_seconds = args.max_lag_minutes * 60 + cooldown_seconds = args.alert_cooldown_hours * 3600 + + try: + rows = fetch_chain_metadata() + except IndexerUnavailableError as exc: + # 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( + f"Envio indexer unavailable: {exc}\nDashboard: {DASHBOARD_URL}", + PROTOCOL, + source="indexer_freshness", + ) + return + + now = int(time.time()) + freshness = collect_freshness(rows, now) + stale = [entry for entry in freshness if entry.is_stale(max_lag_seconds)] + missing = missing_chains(freshness) + if missing: + logger.error("No sync state for %s", ", ".join(chain.name for chain in missing)) + + report_recovered([entry for entry in freshness if entry not in stale and entry.lag_seconds is not None]) + + if not stale and not missing: + logger.info("Indexer is fresh on all %d expected chain(s)", len(freshness)) + return + + stale_to_alert, missing_to_alert = chains_to_alert(stale, missing, now, cooldown_seconds) + if not stale_to_alert and not missing_to_alert: + logger.info("All %d unhealthy chain(s) already alerted within the cooldown window", len(stale) + len(missing)) + return + + send_error_message( + build_alert_message(stale_to_alert, missing_to_alert, max_lag_seconds), + PROTOCOL, + source="indexer_freshness", + ) + for entry in stale_to_alert: + _set_last_alert_timestamp(entry.chain.chain_id, now) + for chain in missing_to_alert: + _set_last_alert_timestamp(chain.chain_id, now) + + +def parse_args() -> argparse.Namespace: + """Parse CLI arguments.""" + parser = argparse.ArgumentParser(description="Alert when the Envio indexer falls behind the chain head") + parser.add_argument( + "--max-lag-minutes", + type=int, + default=int(os.getenv("INDEXER_MAX_LAG_MINUTES", DEFAULT_MAX_LAG_MINUTES)), + help=f"Alert when a chain's newest indexed block is older than this (default: {DEFAULT_MAX_LAG_MINUTES})", + ) + parser.add_argument( + "--alert-cooldown-hours", + type=int, + default=int(os.getenv("INDEXER_ALERT_COOLDOWN_HOURS", DEFAULT_ALERT_COOLDOWN_HOURS)), + help=f"Minimum hours between repeat alerts for the same chain (default: {DEFAULT_ALERT_COOLDOWN_HOURS})", + ) + return parser.parse_args() + + +if __name__ == "__main__": + from utils.runner import run_with_alert + + run_with_alert(main, PROTOCOL) diff --git a/tests/test_indexer_freshness.py b/tests/test_indexer_freshness.py new file mode 100644 index 0000000..599f0d0 --- /dev/null +++ b/tests/test_indexer_freshness.py @@ -0,0 +1,309 @@ +"""Tests for the Envio indexer freshness monitor.""" + +import pytest +import requests + +from protocols.yearn import check_indexer_freshness as freshness +from protocols.yearn.check_indexer_freshness import ChainFreshness, IndexerUnavailableError +from utils.chains import Chain + +NOW = 1_800_000_000 +HOUR = 3600 + +# Last processed block per expected chain, roughly as the live indexer reports them. +INDEXED_BLOCKS: dict[Chain, int] = { + Chain.BASE: 49220190, + Chain.MAINNET: 24150245, + Chain.KATANA: 38465232, + Chain.OPTIMISM: 154815667, + Chain.POLYGON: 91016489, + Chain.ARBITRUM: 488554295, +} + +# Chains the indexer covers that nothing in this repo reads from. +UNMONITORED_ROWS = [ + {"chain_id": 100, "latest_processed_block": 47431537, "block_height": 47431737}, + {"chain_id": 80094, "latest_processed_block": 24107494, "block_height": 24107694}, +] + + +class FakeResponse: + def __init__(self, payload: dict) -> None: + self.payload = payload + + def json(self) -> dict: + return self.payload + + +def _rows(*, omit: tuple[Chain, ...] = (), zero_block: tuple[Chain, ...] = ()) -> list[dict]: + """chain_metadata as the indexer returns it: unsorted, including chains we ignore. + + Args: + omit: Expected chains to leave out entirely, as a dropped indexer config would. + zero_block: Expected chains present but with no processed block. + """ + rows = [ + { + "chain_id": chain.chain_id, + "latest_processed_block": 0 if chain in zero_block else block, + "block_height": block + 200, + } + for chain, block in INDEXED_BLOCKS.items() + if chain not in omit + ] + return rows + UNMONITORED_ROWS + + +@pytest.fixture +def envio_url(monkeypatch: pytest.MonkeyPatch) -> str: + url = "https://indexer.example/v1/graphql" + monkeypatch.setattr(freshness, "ENVIO_GRAPHQL_URL", url) + return url + + +@pytest.fixture +def sent(monkeypatch: pytest.MonkeyPatch) -> list[str]: + """Capture every message the monitor routes to the errors channel.""" + messages: list[str] = [] + monkeypatch.setattr(freshness, "send_error_message", lambda msg, protocol, **kwargs: messages.append(msg)) + return messages + + +def test_collect_freshness_computes_lag_and_sorts_by_chain(monkeypatch: pytest.MonkeyPatch) -> None: + timestamps = {Chain.MAINNET: NOW - 5 * HOUR} + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: timestamps.get(chain, NOW - 120)) + + result = freshness.collect_freshness(_rows(), NOW) + + # Gnosis (100) and Berachain (80094) are indexed but unused here, so they drop out. + assert [entry.chain for entry in result] == [ + Chain.MAINNET, + Chain.OPTIMISM, + Chain.POLYGON, + Chain.BASE, + Chain.ARBITRUM, + Chain.KATANA, + ] + assert result[0].lag_seconds == 5 * HOUR + assert result[0].name == "Mainnet" + assert result[1].lag_seconds == 120 + + +def test_collect_freshness_skips_chains_without_a_processed_block(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: NOW) + + result = freshness.collect_freshness(_rows(zero_block=(Chain.KATANA,)), NOW) + + assert Chain.KATANA not in [entry.chain for entry in result] + + +def test_collect_freshness_marks_lag_unknown_when_rpc_fails(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: None) + + result = freshness.collect_freshness(_rows(), NOW) + + assert all(entry.lag_seconds is None for entry in result) + # Unknown lag must never fire an alert — a broken RPC is not a stale indexer. + assert not any(entry.is_stale(HOUR) for entry in result) + + +def test_is_stale_uses_threshold() -> None: + entry = ChainFreshness(chain=Chain.MAINNET, latest_processed_block=100, lag_seconds=HOUR + 1) + + assert entry.is_stale(HOUR) + assert not entry.is_stale(2 * HOUR) + + +def test_missing_chains_covers_absent_and_unsynced_chains(monkeypatch: pytest.MonkeyPatch) -> None: + """A chain dropped from chain_metadata and one with no processed block both count.""" + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: NOW - 60) + rows = _rows(omit=(Chain.KATANA,), zero_block=(Chain.POLYGON,)) + + missing = freshness.missing_chains(freshness.collect_freshness(rows, NOW)) + + assert missing == [Chain.POLYGON, Chain.KATANA] + + +def test_missing_chains_empty_when_every_expected_chain_reports(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: NOW - 60) + + assert freshness.missing_chains(freshness.collect_freshness(_rows(), NOW)) == [] + + +def test_build_alert_message_lists_lagging_and_missing_chains() -> None: + stale = [ + ChainFreshness(chain=Chain.MAINNET, latest_processed_block=24150245, lag_seconds=205 * 86400), + ChainFreshness(chain=Chain.KATANA, latest_processed_block=38465232, lag_seconds=2 * HOUR + 900), + ] + + message = freshness.build_alert_message(stale, [Chain.BASE], HOUR) + + assert "problem on 3 chain(s)" in message + assert "Mainnet (chain 1): 205d behind, last block 24150245" in message + assert "Katana (chain 747474): 2h 15m behind, last block 38465232" in message + assert "Base (chain 8453): no sync state reported by the indexer" in message + assert "Threshold: 1h" in message + assert freshness.DASHBOARD_URL in message + + +def test_fetch_chain_metadata_returns_rows(monkeypatch: pytest.MonkeyPatch, envio_url: str) -> None: + captured: dict = {} + + def fake_request(method: str, url: str, **kwargs): + captured.update(method=method, url=url, **kwargs) + return FakeResponse({"data": {"chain_metadata": _rows()}}) + + monkeypatch.setattr(freshness, "request_with_retry", fake_request) + + assert freshness.fetch_chain_metadata() == _rows() + assert captured["method"] == "post" + assert captured["url"] == envio_url + + +@pytest.mark.parametrize( + "payload", + [ + {"errors": [{"message": "boom"}]}, + {"data": {"chain_metadata": []}}, + {"data": None}, + ], +) +def test_fetch_chain_metadata_raises_on_bad_payload( + monkeypatch: pytest.MonkeyPatch, envio_url: str, payload: dict +) -> None: + monkeypatch.setattr(freshness, "request_with_retry", lambda *a, **kw: FakeResponse(payload)) + + with pytest.raises(IndexerUnavailableError): + freshness.fetch_chain_metadata() + + +def test_fetch_chain_metadata_raises_when_url_missing(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(freshness, "ENVIO_GRAPHQL_URL", None) + + with pytest.raises(IndexerUnavailableError, match="ENVIO_GRAPHQL_URL"): + freshness.fetch_chain_metadata() + + +def test_chains_to_alert_respects_cooldown() -> None: + stale = [ChainFreshness(chain=Chain.MAINNET, latest_processed_block=1, lag_seconds=2 * HOUR)] + missing = [Chain.KATANA] + + assert freshness.chains_to_alert(stale, missing, NOW, 6 * HOUR) == (stale, missing) + + freshness._set_last_alert_timestamp(Chain.MAINNET.chain_id, NOW) + freshness._set_last_alert_timestamp(Chain.KATANA.chain_id, NOW) + assert freshness.chains_to_alert(stale, missing, NOW + HOUR, 6 * HOUR) == ([], []) + assert freshness.chains_to_alert(stale, missing, NOW + 6 * HOUR, 6 * HOUR) == (stale, missing) + + +def test_chains_to_alert_keeps_chains_independent() -> None: + """One chain inside its cooldown must not suppress another's first alert.""" + stale = [ChainFreshness(chain=Chain.MAINNET, latest_processed_block=1, lag_seconds=2 * HOUR)] + freshness._set_last_alert_timestamp(Chain.MAINNET.chain_id, NOW) + + assert freshness.chains_to_alert(stale, [Chain.BASE], NOW, 6 * HOUR) == ([], [Chain.BASE]) + + +def test_main_alerts_once_per_cooldown_then_reports_recovery( + monkeypatch: pytest.MonkeyPatch, envio_url: str, sent: list[str] +) -> None: + monkeypatch.setattr( + freshness, "request_with_retry", lambda *a, **kw: FakeResponse({"data": {"chain_metadata": _rows()}}) + ) + monkeypatch.setattr("sys.argv", ["check_indexer_freshness.py"]) + + timestamps = {Chain.MAINNET: NOW - 5 * HOUR} + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: timestamps.get(chain, NOW - 120)) + monkeypatch.setattr(freshness.time, "time", lambda: NOW) + + freshness.main() + assert len(sent) == 1 + assert "Mainnet" in sent[0] + assert "Base" not in sent[0] + + # Second run inside the cooldown window stays quiet. + freshness.main() + assert len(sent) == 1 + + # Once mainnet catches up, the recovery note fires and clears the state. + timestamps[Chain.MAINNET] = NOW - 60 + freshness.main() + assert len(sent) == 2 + assert "caught up" in sent[1] + + freshness.main() + assert len(sent) == 2 + + +def test_main_alerts_when_an_expected_chain_is_absent( + monkeypatch: pytest.MonkeyPatch, envio_url: str, sent: list[str] +) -> None: + """Rows present but an expected chain missing must not read as "all fresh".""" + rows = _rows(omit=(Chain.KATANA,)) + monkeypatch.setattr( + freshness, "request_with_retry", lambda *a, **kw: FakeResponse({"data": {"chain_metadata": rows}}) + ) + monkeypatch.setattr("sys.argv", ["check_indexer_freshness.py"]) + # Every chain the indexer *does* report is perfectly fresh. + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: NOW - 60) + monkeypatch.setattr(freshness.time, "time", lambda: NOW) + + freshness.main() + + assert len(sent) == 1 + assert "Katana (chain 747474): no sync state reported by the indexer" in sent[0] + + +def test_main_stays_quiet_when_every_expected_chain_is_fresh( + monkeypatch: pytest.MonkeyPatch, envio_url: str, sent: list[str] +) -> None: + monkeypatch.setattr( + freshness, "request_with_retry", lambda *a, **kw: FakeResponse({"data": {"chain_metadata": _rows()}}) + ) + monkeypatch.setattr("sys.argv", ["check_indexer_freshness.py"]) + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: NOW - 60) + monkeypatch.setattr(freshness.time, "time", lambda: NOW) + + freshness.main() + + assert sent == [] + + +def test_main_alerts_when_indexer_is_unreachable( + monkeypatch: pytest.MonkeyPatch, envio_url: str, sent: list[str] +) -> None: + monkeypatch.setattr("sys.argv", ["check_indexer_freshness.py"]) + monkeypatch.setattr(freshness, "request_with_retry", lambda *a, **kw: FakeResponse({"errors": ["down"]})) + + freshness.main() + + assert len(sent) == 1 + assert "Envio indexer unavailable" in sent[0] + + +@pytest.mark.parametrize( + "error", + [ + # request_with_retry exhausts its retries on 5xx, then raises HTTPError. + requests.HTTPError("502 Server Error: Bad Gateway"), + requests.ConnectionError("connection refused"), + requests.Timeout("read timeout"), + # Hasura down behind a proxy answers 200 with an HTML error page. + ValueError("Expecting value: line 1 column 1 (char 0)"), + ], +) +def test_main_alerts_on_transport_failure( + monkeypatch: pytest.MonkeyPatch, envio_url: str, sent: list[str], error: Exception +) -> None: + """Every way the endpoint can fail still produces a Telegram alert.""" + monkeypatch.setattr("sys.argv", ["check_indexer_freshness.py"]) + + def fail(*args, **kwargs): + raise error + + monkeypatch.setattr(freshness, "request_with_retry", fail) + + freshness.main() + + assert len(sent) == 1 + assert "Envio indexer unavailable" in sent[0] diff --git a/utils/formatting.py b/utils/formatting.py index 9d1ebe8..159a1ca 100644 --- a/utils/formatting.py +++ b/utils/formatting.py @@ -20,3 +20,30 @@ def format_usd(number: float) -> str: def format_token_amount(raw: int, decimals: int) -> float: """Convert a raw token amount to a human-readable float.""" return raw / (10**decimals) + + +def format_duration(seconds: int) -> str: + """Format a duration in seconds as a compact human-readable string. + + Minutes are dropped once the duration spans days, so long durations stay + readable (e.g. "12d 3h" rather than "12d 3h 47m"). + + Args: + seconds: Duration in seconds. Zero or negative renders as "now". + + Returns: + A string like "now", "45s", "12m", "3h 5m" or "12d 3h". + """ + if seconds <= 0: + return "now" + days = seconds // 86400 + hours = (seconds % 86400) // 3600 + minutes = (seconds % 3600) // 60 + parts: list[str] = [] + if days: + parts.append(f"{days}d") + if hours: + parts.append(f"{hours}h") + if minutes and not days: + parts.append(f"{minutes}m") + return " ".join(parts) if parts else f"{seconds}s" diff --git a/utils/web3_wrapper.py b/utils/web3_wrapper.py index 01af3b6..8a19064 100644 --- a/utils/web3_wrapper.py +++ b/utils/web3_wrapper.py @@ -9,7 +9,7 @@ from web3.contract import Contract from web3.exceptions import ProviderConnectionError from web3.providers.rpc import HTTPProvider -from web3.types import RPCResponse +from web3.types import RPCEndpoint, RPCResponse from utils.logger import get_logger @@ -175,6 +175,16 @@ def get_contract(self, address: str, abi: List[Dict]) -> Contract: """Get contract instance""" return self.w3.eth.contract(address=address, abi=abi) + def make_request(self, method: str, params: List[Any]) -> RPCResponse: + """Send a raw JSON-RPC request, bypassing web3's response formatters. + + Use this when web3's typed accessors reject an otherwise valid response. + `eth.get_block` on Polygon or pre-Bedrock Optimism, for example, raises + on the 97-byte PoA `extraData` field unless the PoA middleware is + injected. Provider rotation and retries still apply. + """ + return self.w3.provider.make_request(RPCEndpoint(method), params) + def batch_requests(self): return self.w3.batch_requests()