From de86834b7ffee515a7db755de0e9f8e072e87fc2 Mon Sep 17 00:00:00 2001 From: spalen0 Date: Tue, 28 Jul 2026 11:37:09 +0200 Subject: [PATCH 1/3] feat(yearn): alert when the Envio indexer falls behind MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The monitors that read from the Envio indexer (large flows, timelock alerts, 3jane borrower watch) go quiet rather than loud when it stalls: GraphQL keeps answering, it just stops returning new rows, so a 24h+ outage is indistinguishable from a quiet day. Add an hourly check that reads chain_metadata and resolves the wall-clock timestamp of each chain's latest_processed_block over JSON-RPC, alerting the errors channel when a chain is more than 60 minutes stale or the endpoint itself is unreachable. 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 comparing those two fields would report a stalled indexer as zero blocks behind (same trap documented in the indexer's own monitoring dashboard). Each chain alerts on entering staleness, then at most once per cooldown window (default 6h), then once on recovery — a re-sync can run for days, and today Mainnet is ~206d behind and Katana ~6h. Also move the duplicated format_duration helper into utils/formatting. Co-Authored-By: Claude Opus 5 (1M context) --- .env.example | 8 + automation/jobs.yaml | 4 + monitoring.yaml | 3 + protocols/3jane/main.py | 18 +- protocols/yearn/README.md | 39 +++ protocols/yearn/check_indexer_freshness.py | 334 +++++++++++++++++++++ tests/test_indexer_freshness.py | 180 +++++++++++ utils/formatting.py | 27 ++ 8 files changed, 596 insertions(+), 17 deletions(-) create mode 100644 protocols/yearn/check_indexer_freshness.py create mode 100644 tests/test_indexer_freshness.py diff --git a/.env.example b/.env.example index 8ebb4af5..0012ba58 100644 --- a/.env.example +++ b/.env.example @@ -17,6 +17,14 @@ 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 +# The indexer also covers Gnosis and Berachain, which have no Chain enum member +# here; set these to replace the public RPC fallbacks used for those two. +# PROVIDER_URL_GNOSIS=https://rpc.gnosischain.com +# PROVIDER_URL_BERACHAIN=https://rpc.berachain.com + # 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 710b223b..da74d104 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 7a26fb64..1523f90f 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 3bcd9385..691ce064 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 3616221a..67aac67d 100644 --- a/protocols/yearn/README.md +++ b/protocols/yearn/README.md @@ -239,3 +239,42 @@ 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 every indexed chain's `latest_processed_block`. +2. Fetches that block's timestamp over JSON-RPC and compares it to wall-clock time. +3. Alerts when a chain's newest indexed block is older than `--max-lag-minutes` (default `60`). + +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/). + +RPCs resolve from this repo's `PROVIDER_URL_*` variables. The indexer also covers Gnosis and Berachain, which have no `Chain` enum member here, so those fall back to a public endpoint (override with `PROVIDER_URL_GNOSIS` / `PROVIDER_URL_BERACHAIN`). A chain with no reachable RPC is logged and skipped rather than alerted on — a broken RPC is not a stale indexer. + +### Alerts + +All alerts go to the errors channel (`TELEGRAM_*_ERRORS`) labelled `[yearn]`, alongside the other operational diagnostics: + +- **Stale chains** — one message listing every lagging chain with its lag and last indexed block. +- **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 staleness and then at most once per `--alert-cooldown-hours` (default `6`) instead of every hourly run. The last-alert timestamp per chain 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 00000000..41851c25 --- /dev/null +++ b/protocols/yearn/check_indexer_freshness.py @@ -0,0 +1,334 @@ +#!/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 indexed chain, +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. + +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 + +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_" + +# Mirrors the chain list in the indexer's monitoring dashboard (apps/monitoring +# in chain-events/yearn-indexing-test). Unlisted chains still get checked, they +# just render as "Chain ". +CHAIN_NAMES: dict[int, str] = { + 1: "Ethereum", + 10: "Optimism", + 100: "Gnosis", + 137: "Polygon", + 8453: "Base", + 42161: "Arbitrum", + 80094: "Berachain", + 747474: "Katana", +} + +# The indexer covers chains this repo has no `Chain` enum member (and therefore +# no PROVIDER_URL_*) for. A public endpoint is enough for the single +# `eth_getBlockByNumber` per run; override with PROVIDER_URL_ when a +# dedicated provider is available. +FALLBACK_RPC_URLS: dict[int, str] = { + 100: os.getenv("PROVIDER_URL_GNOSIS", "https://rpc.gnosischain.com"), + 80094: os.getenv("PROVIDER_URL_BERACHAIN", "https://rpc.berachain.com"), +} + +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_id: int + latest_processed_block: int + # None when no RPC could resolve the block timestamp, i.e. lag is unknown. + lag_seconds: int | None + + @property + def name(self) -> str: + """Human-readable chain name, falling back to the raw id.""" + return CHAIN_NAMES.get(self.chain_id, f"Chain {self.chain_id}") + + 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 _rpc_url(chain_id: int) -> str | None: + """Resolve an RPC URL for a chain, preferring this repo's configured providers.""" + try: + chain = Chain.from_chain_id(chain_id) + except ValueError: + return FALLBACK_RPC_URLS.get(chain_id) + + for env_key in (f"PROVIDER_URL_{chain.name}", *(f"PROVIDER_URL_{chain.name}_{i}" for i in range(1, 4))): + url = os.getenv(env_key) + if url: + return url + return FALLBACK_RPC_URLS.get(chain_id) + + +def fetch_block_timestamp(chain_id: int, block_number: int) -> int | None: + """Fetch the unix timestamp of a block via JSON-RPC. + + Args: + chain_id: Chain the block belongs to. + block_number: Block to look up. + + Returns: + The block timestamp in seconds, or None when no RPC is configured for the + chain or the call failed — an unreachable RPC must not mask the other + chains' results. + """ + url = _rpc_url(chain_id) + if not url: + logger.warning("No RPC configured for chain %d, skipping freshness check", chain_id) + return None + + body = {"jsonrpc": "2.0", "id": 1, "method": "eth_getBlockByNumber", "params": [hex(block_number), False]} + try: + response = request_with_retry("post", url, json=body) + block = (response.json() or {}).get("result") + if not block or "timestamp" not in block: + logger.warning("RPC for chain %d returned no block %d", chain_id, block_number) + return None + return int(block["timestamp"], 16) + except (requests.RequestException, ValueError, TypeError) as exc: + logger.warning("Failed to fetch block %d timestamp on chain %d: %s", block_number, chain_id, exc) + return None + + +def collect_freshness(rows: list[dict], now: int) -> list[ChainFreshness]: + """Resolve how far behind wall-clock time each indexed chain is. + + Args: + rows: `chain_metadata` rows from the indexer. + now: Current unix timestamp. + + Returns: + One ChainFreshness per row, sorted by chain id. + """ + freshness: list[ChainFreshness] = [] + for row in rows: + chain_id = int(row["chain_id"]) + latest_block = int(row.get("latest_processed_block") or 0) + if latest_block <= 0: + # A chain that has never processed a block is mid-backfill, not stale. + logger.warning("Chain %d has no processed block yet, skipping", chain_id) + continue + block_timestamp = fetch_block_timestamp(chain_id, 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("Chain %d: block %d, lag %s", chain_id, latest_block, lag_text) + freshness.append(ChainFreshness(chain_id=chain_id, latest_processed_block=latest_block, lag_seconds=lag)) + return sorted(freshness, key=lambda f: f.chain_id) + + +def build_stale_message(stale: list[ChainFreshness], max_lag_seconds: int) -> str: + """Build the plain-text alert body listing every lagging chain.""" + lines = [ + f"Envio indexer is behind on {len(stale)} chain(s) — events may be missing from monitoring alerts.", + "", + ] + for chain in stale: + lag = format_duration(chain.lag_seconds or 0) + lines.append( + f"- {chain.name} (chain {chain.chain_id}): {lag} behind, last block {chain.latest_processed_block}" + ) + 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 chains_to_alert(stale: list[ChainFreshness], now: int, cooldown_seconds: int) -> list[ChainFreshness]: + """Filter stale chains down to those outside their re-alert cooldown. + + Args: + stale: Chains currently past the lag threshold. + now: Current unix timestamp. + cooldown_seconds: Minimum gap between two alerts for the same chain. + + Returns: + The chains that should alert on this run. + """ + return [chain for chain in stale if now - _last_alert_timestamp(chain.chain_id) >= 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_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_id, 0) + + +def main() -> None: + """Check indexer freshness for every indexed chain and alert on stale ones.""" + 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 = [chain for chain in freshness if chain.is_stale(max_lag_seconds)] + + report_recovered([chain for chain in freshness if chain not in stale and chain.lag_seconds is not None]) + + if not stale: + logger.info("Indexer is fresh on all %d chain(s)", len(freshness)) + return + + to_alert = chains_to_alert(stale, now, cooldown_seconds) + if not to_alert: + logger.info("All %d stale chain(s) already alerted within the cooldown window", len(stale)) + return + + send_error_message(build_stale_message(to_alert, max_lag_seconds), PROTOCOL, source="indexer_freshness") + for chain in 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 00000000..ffacd62f --- /dev/null +++ b/tests/test_indexer_freshness.py @@ -0,0 +1,180 @@ +"""Tests for the Envio indexer freshness monitor.""" + +import pytest + +from protocols.yearn import check_indexer_freshness as freshness +from protocols.yearn.check_indexer_freshness import ChainFreshness, IndexerUnavailableError + +NOW = 1_800_000_000 +HOUR = 3600 + + +class FakeResponse: + def __init__(self, payload: dict) -> None: + self.payload = payload + + def json(self) -> dict: + return self.payload + + +def _rows() -> list[dict]: + return [ + {"chain_id": 8453, "latest_processed_block": 49220190, "block_height": 49220390}, + {"chain_id": 1, "latest_processed_block": 24150245, "block_height": 25626800}, + ] + + +@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 = {1: NOW - 5 * HOUR, 8453: NOW - 120} + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain_id, block: timestamps[chain_id]) + + result = freshness.collect_freshness(_rows(), NOW) + + assert [c.chain_id for c in result] == [1, 8453] + assert result[0].lag_seconds == 5 * HOUR + assert result[0].name == "Ethereum" + 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_id, block: NOW) + rows = [{"chain_id": 100, "latest_processed_block": 0, "block_height": 0}] + + assert freshness.collect_freshness(rows, NOW) == [] + + +def test_collect_freshness_marks_lag_unknown_when_rpc_fails(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain_id, block: None) + + result = freshness.collect_freshness(_rows(), NOW) + + assert all(chain.lag_seconds is None for chain in result) + # Unknown lag must never fire an alert — a broken RPC is not a stale indexer. + assert not any(chain.is_stale(HOUR) for chain in result) + + +def test_is_stale_uses_threshold() -> None: + chain = ChainFreshness(chain_id=1, latest_processed_block=100, lag_seconds=HOUR + 1) + + assert chain.is_stale(HOUR) + assert not chain.is_stale(2 * HOUR) + + +def test_build_stale_message_lists_every_lagging_chain() -> None: + stale = [ + ChainFreshness(chain_id=1, latest_processed_block=24150245, lag_seconds=205 * 86400), + ChainFreshness(chain_id=747474, latest_processed_block=38465232, lag_seconds=2 * HOUR + 900), + ] + + message = freshness.build_stale_message(stale, HOUR) + + assert "Ethereum (chain 1): 205d behind, last block 24150245" in message + assert "Katana (chain 747474): 2h 15m behind, last block 38465232" 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_id=1, latest_processed_block=1, lag_seconds=2 * HOUR)] + + assert freshness.chains_to_alert(stale, NOW, 6 * HOUR) == stale + + freshness._set_last_alert_timestamp(1, NOW) + assert freshness.chains_to_alert(stale, NOW + HOUR, 6 * HOUR) == [] + assert freshness.chains_to_alert(stale, NOW + 6 * HOUR, 6 * HOUR) == stale + + +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"]) + + stale_timestamps = {1: NOW - 5 * HOUR, 8453: NOW - 120} + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain_id, block: stale_timestamps[chain_id]) + monkeypatch.setattr(freshness.time, "time", lambda: NOW) + + freshness.main() + assert len(sent) == 1 + assert "Ethereum" 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. + stale_timestamps[1] = 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_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] diff --git a/utils/formatting.py b/utils/formatting.py index 9d1ebe84..159a1ca8 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" From f275844cc24ae035e1cafec99482fab156d40faf Mon Sep 17 00:00:00 2001 From: spalen0 Date: Tue, 28 Jul 2026 11:53:50 +0200 Subject: [PATCH 2/3] refactor(yearn): only check indexer freshness for chains we use MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drop Gnosis and Berachain from the freshness check — the indexer covers them but nothing in this repo reads their events, so they were alerting on data we never consume. That removes the public-RPC fallbacks too, so the check now runs entirely through ChainManager on the six Chain enum members. Fetch the block via a raw eth_getBlockByNumber rather than eth.get_block: web3 rejects the 97-byte PoA extraData on Polygon and pre-Bedrock Optimism blocks, and old blocks are exactly what a lagging indexer points at. Add Web3Client.make_request for that, keeping provider rotation. Co-Authored-By: Claude Opus 5 (1M context) --- .env.example | 4 - protocols/yearn/README.md | 6 +- protocols/yearn/check_indexer_freshness.py | 122 ++++++++------------- tests/test_indexer_freshness.py | 66 ++++++++--- utils/web3_wrapper.py | 12 +- 5 files changed, 109 insertions(+), 101 deletions(-) diff --git a/.env.example b/.env.example index 0012ba58..c0b3a9ab 100644 --- a/.env.example +++ b/.env.example @@ -20,10 +20,6 @@ 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 -# The indexer also covers Gnosis and Berachain, which have no Chain enum member -# here; set these to replace the public RPC fallbacks used for those two. -# PROVIDER_URL_GNOSIS=https://rpc.gnosischain.com -# PROVIDER_URL_BERACHAIN=https://rpc.berachain.com # Telegram API credentials TELEGRAM_BOT_TOKEN_DEFAULT=your-default-bot-token diff --git a/protocols/yearn/README.md b/protocols/yearn/README.md index 67aac67d..9409f462 100644 --- a/protocols/yearn/README.md +++ b/protocols/yearn/README.md @@ -250,13 +250,13 @@ An indexer stall is invisible to the monitors that depend on it: GraphQL keeps a ### How It Works -1. Queries `chain_metadata` at `ENVIO_GRAPHQL_URL` for every indexed chain's `latest_processed_block`. -2. Fetches that block's timestamp over JSON-RPC and compares it to wall-clock time. +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`). 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/). -RPCs resolve from this repo's `PROVIDER_URL_*` variables. The indexer also covers Gnosis and Berachain, which have no `Chain` enum member here, so those fall back to a public endpoint (override with `PROVIDER_URL_GNOSIS` / `PROVIDER_URL_BERACHAIN`). A chain with no reachable RPC is logged and skipped rather than alerted on — a broken RPC is not a stale indexer. +Only the six chains in the `Chain` enum are checked (Mainnet, Optimism, Base, Arbitrum, Polygon, Katana). 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 diff --git a/protocols/yearn/check_indexer_freshness.py b/protocols/yearn/check_indexer_freshness.py index 41851c25..159579f0 100644 --- a/protocols/yearn/check_indexer_freshness.py +++ b/protocols/yearn/check_indexer_freshness.py @@ -6,10 +6,10 @@ 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 indexed chain, -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. +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. 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 @@ -35,6 +35,7 @@ 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() @@ -59,29 +60,6 @@ # Keyed per chain so one lagging chain can't suppress an alert for another. CACHE_KEY_LAST_ALERT_PREFIX = "YEARN_INDEXER_STALE_ALERT_" -# Mirrors the chain list in the indexer's monitoring dashboard (apps/monitoring -# in chain-events/yearn-indexing-test). Unlisted chains still get checked, they -# just render as "Chain ". -CHAIN_NAMES: dict[int, str] = { - 1: "Ethereum", - 10: "Optimism", - 100: "Gnosis", - 137: "Polygon", - 8453: "Base", - 42161: "Arbitrum", - 80094: "Berachain", - 747474: "Katana", -} - -# The indexer covers chains this repo has no `Chain` enum member (and therefore -# no PROVIDER_URL_*) for. A public endpoint is enough for the single -# `eth_getBlockByNumber` per run; override with PROVIDER_URL_ when a -# dedicated provider is available. -FALLBACK_RPC_URLS: dict[int, str] = { - 100: os.getenv("PROVIDER_URL_GNOSIS", "https://rpc.gnosischain.com"), - 80094: os.getenv("PROVIDER_URL_BERACHAIN", "https://rpc.berachain.com"), -} - CHAIN_METADATA_QUERY = """ { chain_metadata { @@ -101,15 +79,15 @@ class IndexerUnavailableError(Exception): class ChainFreshness: """Freshness of a single indexed chain.""" - chain_id: int + chain: Chain latest_processed_block: int - # None when no RPC could resolve the block timestamp, i.e. lag is unknown. + # 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, falling back to the raw id.""" - return CHAIN_NAMES.get(self.chain_id, f"Chain {self.chain_id}") + """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.""" @@ -147,74 +125,64 @@ def fetch_chain_metadata() -> list[dict]: return rows -def _rpc_url(chain_id: int) -> str | None: - """Resolve an RPC URL for a chain, preferring this repo's configured providers.""" - try: - chain = Chain.from_chain_id(chain_id) - except ValueError: - return FALLBACK_RPC_URLS.get(chain_id) - - for env_key in (f"PROVIDER_URL_{chain.name}", *(f"PROVIDER_URL_{chain.name}_{i}" for i in range(1, 4))): - url = os.getenv(env_key) - if url: - return url - return FALLBACK_RPC_URLS.get(chain_id) - - -def fetch_block_timestamp(chain_id: int, block_number: int) -> int | None: - """Fetch the unix timestamp of a block via JSON-RPC. +def fetch_block_timestamp(chain: Chain, block_number: int) -> int | None: + """Fetch the unix timestamp of a block. Args: - chain_id: Chain the block belongs to. + chain: Chain the block belongs to. block_number: Block to look up. Returns: - The block timestamp in seconds, or None when no RPC is configured for the - chain or the call failed — an unreachable RPC must not mask the other - chains' results. - """ - url = _rpc_url(chain_id) - if not url: - logger.warning("No RPC configured for chain %d, skipping freshness check", chain_id) - return None + The block timestamp in seconds, or None when the lookup failed — one + unreachable RPC must not mask the other chains' results. - body = {"jsonrpc": "2.0", "id": 1, "method": "eth_getBlockByNumber", "params": [hex(block_number), False]} + 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: - response = request_with_retry("post", url, json=body) - block = (response.json() or {}).get("result") - if not block or "timestamp" not in block: - logger.warning("RPC for chain %d returned no block %d", chain_id, block_number) - return None - return int(block["timestamp"], 16) - except (requests.RequestException, ValueError, TypeError) as exc: - logger.warning("Failed to fetch block %d timestamp on chain %d: %s", block_number, chain_id, exc) + 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 indexed chain is. + """Resolve how far behind wall-clock time each monitored chain is. + + The indexer covers chains this repo doesn't read from (Gnosis, Berachain). + They have no `Chain` member and nothing here consumes their events, so they + are skipped rather than alerted on. Args: rows: `chain_metadata` rows from the indexer. now: Current unix timestamp. Returns: - One ChainFreshness per row, sorted by chain id. + One ChainFreshness per monitored chain, sorted by chain id. """ freshness: list[ChainFreshness] = [] for row in rows: chain_id = int(row["chain_id"]) + try: + chain = Chain.from_chain_id(chain_id) + except ValueError: + 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: # A chain that has never processed a block is mid-backfill, not stale. - logger.warning("Chain %d has no processed block yet, skipping", chain_id) + logger.warning("%s has no processed block yet, skipping", chain.name) continue - block_timestamp = fetch_block_timestamp(chain_id, latest_block) + 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("Chain %d: block %d, lag %s", chain_id, latest_block, lag_text) - freshness.append(ChainFreshness(chain_id=chain_id, latest_processed_block=latest_block, lag_seconds=lag)) - return sorted(freshness, key=lambda f: f.chain_id) + 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 build_stale_message(stale: list[ChainFreshness], max_lag_seconds: int) -> str: @@ -226,7 +194,7 @@ def build_stale_message(stale: list[ChainFreshness], max_lag_seconds: int) -> st for chain in stale: lag = format_duration(chain.lag_seconds or 0) lines.append( - f"- {chain.name} (chain {chain.chain_id}): {lag} behind, last block {chain.latest_processed_block}" + f"- {chain.name} (chain {chain.chain.chain_id}): {lag} behind, last block {chain.latest_processed_block}" ) lines += [ "", @@ -257,18 +225,18 @@ def chains_to_alert(stale: list[ChainFreshness], now: int, cooldown_seconds: int Returns: The chains that should alert on this run. """ - return [chain for chain in stale if now - _last_alert_timestamp(chain.chain_id) >= cooldown_seconds] + return [chain for chain in stale if now - _last_alert_timestamp(chain.chain.chain_id) >= 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_id) > 0] + 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_id, 0) + _set_last_alert_timestamp(chain.chain.chain_id, 0) def main() -> None: @@ -307,7 +275,7 @@ def main() -> None: send_error_message(build_stale_message(to_alert, max_lag_seconds), PROTOCOL, source="indexer_freshness") for chain in to_alert: - _set_last_alert_timestamp(chain.chain_id, now) + _set_last_alert_timestamp(chain.chain.chain_id, now) def parse_args() -> argparse.Namespace: diff --git a/tests/test_indexer_freshness.py b/tests/test_indexer_freshness.py index ffacd62f..5f611a3c 100644 --- a/tests/test_indexer_freshness.py +++ b/tests/test_indexer_freshness.py @@ -1,9 +1,11 @@ """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 @@ -18,9 +20,12 @@ def json(self) -> dict: def _rows() -> list[dict]: + """chain_metadata as the indexer returns it: unsorted, including chains we ignore.""" return [ {"chain_id": 8453, "latest_processed_block": 49220190, "block_height": 49220390}, + {"chain_id": 80094, "latest_processed_block": 24107494, "block_height": 24107694}, {"chain_id": 1, "latest_processed_block": 24150245, "block_height": 25626800}, + {"chain_id": 100, "latest_processed_block": 47431537, "block_height": 47431737}, ] @@ -40,26 +45,27 @@ def sent(monkeypatch: pytest.MonkeyPatch) -> list[str]: def test_collect_freshness_computes_lag_and_sorts_by_chain(monkeypatch: pytest.MonkeyPatch) -> None: - timestamps = {1: NOW - 5 * HOUR, 8453: NOW - 120} - monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain_id, block: timestamps[chain_id]) + timestamps = {Chain.MAINNET: NOW - 5 * HOUR, Chain.BASE: NOW - 120} + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: timestamps[chain]) result = freshness.collect_freshness(_rows(), NOW) - assert [c.chain_id for c in result] == [1, 8453] + # Gnosis (100) and Berachain (80094) are indexed but unused here, so they drop out. + assert [c.chain for c in result] == [Chain.MAINNET, Chain.BASE] assert result[0].lag_seconds == 5 * HOUR - assert result[0].name == "Ethereum" + 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_id, block: NOW) - rows = [{"chain_id": 100, "latest_processed_block": 0, "block_height": 0}] + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: NOW) + rows = [{"chain_id": 1, "latest_processed_block": 0, "block_height": 0}] assert freshness.collect_freshness(rows, NOW) == [] def test_collect_freshness_marks_lag_unknown_when_rpc_fails(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain_id, block: None) + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: None) result = freshness.collect_freshness(_rows(), NOW) @@ -69,7 +75,7 @@ def test_collect_freshness_marks_lag_unknown_when_rpc_fails(monkeypatch: pytest. def test_is_stale_uses_threshold() -> None: - chain = ChainFreshness(chain_id=1, latest_processed_block=100, lag_seconds=HOUR + 1) + chain = ChainFreshness(chain=Chain.MAINNET, latest_processed_block=100, lag_seconds=HOUR + 1) assert chain.is_stale(HOUR) assert not chain.is_stale(2 * HOUR) @@ -77,13 +83,13 @@ def test_is_stale_uses_threshold() -> None: def test_build_stale_message_lists_every_lagging_chain() -> None: stale = [ - ChainFreshness(chain_id=1, latest_processed_block=24150245, lag_seconds=205 * 86400), - ChainFreshness(chain_id=747474, latest_processed_block=38465232, lag_seconds=2 * HOUR + 900), + 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_stale_message(stale, HOUR) - assert "Ethereum (chain 1): 205d behind, last block 24150245" 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 "Threshold: 1h" in message assert freshness.DASHBOARD_URL in message @@ -128,7 +134,7 @@ def test_fetch_chain_metadata_raises_when_url_missing(monkeypatch: pytest.Monkey def test_chains_to_alert_respects_cooldown() -> None: - stale = [ChainFreshness(chain_id=1, latest_processed_block=1, lag_seconds=2 * HOUR)] + stale = [ChainFreshness(chain=Chain.MAINNET, latest_processed_block=1, lag_seconds=2 * HOUR)] assert freshness.chains_to_alert(stale, NOW, 6 * HOUR) == stale @@ -145,13 +151,13 @@ def test_main_alerts_once_per_cooldown_then_reports_recovery( ) monkeypatch.setattr("sys.argv", ["check_indexer_freshness.py"]) - stale_timestamps = {1: NOW - 5 * HOUR, 8453: NOW - 120} - monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain_id, block: stale_timestamps[chain_id]) + stale_timestamps = {Chain.MAINNET: NOW - 5 * HOUR, Chain.BASE: NOW - 120} + monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: stale_timestamps[chain]) monkeypatch.setattr(freshness.time, "time", lambda: NOW) freshness.main() assert len(sent) == 1 - assert "Ethereum" in sent[0] + assert "Mainnet" in sent[0] assert "Base" not in sent[0] # Second run inside the cooldown window stays quiet. @@ -159,7 +165,7 @@ def test_main_alerts_once_per_cooldown_then_reports_recovery( assert len(sent) == 1 # Once mainnet catches up, the recovery note fires and clears the state. - stale_timestamps[1] = NOW - 60 + stale_timestamps[Chain.MAINNET] = NOW - 60 freshness.main() assert len(sent) == 2 assert "caught up" in sent[1] @@ -178,3 +184,31 @@ def test_main_alerts_when_indexer_is_unreachable( 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/web3_wrapper.py b/utils/web3_wrapper.py index 01af3b6e..8a190643 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() From ae4c957d4d85515feb38d3c7eb7466ad68858578 Mon Sep 17 00:00:00 2001 From: spalen0 Date: Tue, 28 Jul 2026 12:29:36 +0200 Subject: [PATCH 3/3] fix(yearn): alert when an expected chain vanishes from chain_metadata MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An empty result set is not good news. If a chain drops out of the indexer's config, or returns from a restart with no processed block, it simply stops appearing in chain_metadata — collect_freshness produced no entry, `not stale` was true, and the job logged the indexer as fresh while that chain's monitors sat blind. Add EXPECTED_CHAINS as the authority on what must be present and alert on anything absent from it, covering both a missing row and a row with no processed block. Missing chains share the per-chain cooldown and the same alert message as lagging ones. EXPECTED_CHAINS is spelled out rather than derived from the Chain enum 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. Co-Authored-By: Claude Opus 5 (1M context) --- protocols/yearn/README.md | 10 +- protocols/yearn/check_indexer_freshness.py | 119 ++++++++++++----- tests/test_indexer_freshness.py | 147 +++++++++++++++++---- 3 files changed, 215 insertions(+), 61 deletions(-) diff --git a/protocols/yearn/README.md b/protocols/yearn/README.md index 9409f462..d26a4f25 100644 --- a/protocols/yearn/README.md +++ b/protocols/yearn/README.md @@ -252,21 +252,23 @@ An indexer stall is invisible to the monitors that depend on it: GraphQL keeps a 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`). +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/). -Only the six chains in the `Chain` enum are checked (Mainnet, Optimism, Base, Arbitrum, Polygon, Katana). 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. +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 chains** — one message listing every lagging chain with its lag and last indexed block. +- **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 staleness and then at most once per `--alert-cooldown-hours` (default `6`) instead of every hourly run. The last-alert timestamp per chain is cached under `YEARN_INDEXER_STALE_ALERT_`. +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 diff --git a/protocols/yearn/check_indexer_freshness.py b/protocols/yearn/check_indexer_freshness.py index 159579f0..9d3c3c68 100644 --- a/protocols/yearn/check_indexer_freshness.py +++ b/protocols/yearn/check_indexer_freshness.py @@ -9,7 +9,9 @@ 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. +(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 @@ -60,6 +62,25 @@ # 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 { @@ -151,31 +172,32 @@ def fetch_block_timestamp(chain: Chain, block_number: int) -> int | None: def collect_freshness(rows: list[dict], now: int) -> list[ChainFreshness]: - """Resolve how far behind wall-clock time each monitored chain is. + """Resolve how far behind wall-clock time each expected chain is. The indexer covers chains this repo doesn't read from (Gnosis, Berachain). - They have no `Chain` member and nothing here consumes their events, so they - are skipped rather than alerted on. + 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 monitored chain, sorted by chain id. + 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"]) - try: - chain = Chain.from_chain_id(chain_id) - except ValueError: + 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: - # A chain that has never processed a block is mid-backfill, not stale. - logger.warning("%s has no processed block yet, skipping", chain.name) + # 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 @@ -185,17 +207,32 @@ def collect_freshness(rows: list[dict], now: int) -> list[ChainFreshness]: return sorted(freshness, key=lambda f: f.chain.chain_id) -def build_stale_message(stale: list[ChainFreshness], max_lag_seconds: int) -> str: - """Build the plain-text alert body listing every lagging chain.""" +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 is behind on {len(stale)} chain(s) — events may be missing from monitoring alerts.", + f"Envio indexer problem on {affected} chain(s) — events may be missing from monitoring alerts.", "", ] - for chain in stale: - lag = format_duration(chain.lag_seconds or 0) + for entry in stale: + lag = format_duration(entry.lag_seconds or 0) lines.append( - f"- {chain.name} (chain {chain.chain.chain_id}): {lag} behind, last block {chain.latest_processed_block}" + 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)}", @@ -214,18 +251,29 @@ def _set_last_alert_timestamp(chain_id: int, timestamp: int) -> None: write_last_value_to_file(cache_filename, f"{CACHE_KEY_LAST_ALERT_PREFIX}{chain_id}", timestamp) -def chains_to_alert(stale: list[ChainFreshness], now: int, cooldown_seconds: int) -> list[ChainFreshness]: - """Filter stale chains down to those outside their re-alert cooldown. +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 chains that should alert on this run. + The stale and missing chains that should alert on this run. """ - return [chain for chain in stale if now - _last_alert_timestamp(chain.chain.chain_id) >= cooldown_seconds] + 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: @@ -240,7 +288,7 @@ def report_recovered(fresh: list[ChainFreshness]) -> None: def main() -> None: - """Check indexer freshness for every indexed chain and alert on stale ones.""" + """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 @@ -260,22 +308,31 @@ def main() -> None: now = int(time.time()) freshness = collect_freshness(rows, now) - stale = [chain for chain in freshness if chain.is_stale(max_lag_seconds)] + 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([chain for chain in freshness if chain not in stale and chain.lag_seconds is not None]) + report_recovered([entry for entry in freshness if entry not in stale and entry.lag_seconds is not None]) - if not stale: - logger.info("Indexer is fresh on all %d chain(s)", len(freshness)) + if not stale and not missing: + logger.info("Indexer is fresh on all %d expected chain(s)", len(freshness)) return - to_alert = chains_to_alert(stale, now, cooldown_seconds) - if not to_alert: - logger.info("All %d stale chain(s) already alerted within the cooldown window", len(stale)) + 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_stale_message(to_alert, max_lag_seconds), PROTOCOL, source="indexer_freshness") - for chain in to_alert: - _set_last_alert_timestamp(chain.chain.chain_id, now) + 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: diff --git a/tests/test_indexer_freshness.py b/tests/test_indexer_freshness.py index 5f611a3c..599f0d08 100644 --- a/tests/test_indexer_freshness.py +++ b/tests/test_indexer_freshness.py @@ -10,6 +10,22 @@ 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: @@ -19,14 +35,23 @@ def json(self) -> dict: return self.payload -def _rows() -> list[dict]: - """chain_metadata as the indexer returns it: unsorted, including chains we ignore.""" - return [ - {"chain_id": 8453, "latest_processed_block": 49220190, "block_height": 49220390}, - {"chain_id": 80094, "latest_processed_block": 24107494, "block_height": 24107694}, - {"chain_id": 1, "latest_processed_block": 24150245, "block_height": 25626800}, - {"chain_id": 100, "latest_processed_block": 47431537, "block_height": 47431737}, +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 @@ -45,13 +70,20 @@ def sent(monkeypatch: pytest.MonkeyPatch) -> list[str]: def test_collect_freshness_computes_lag_and_sorts_by_chain(monkeypatch: pytest.MonkeyPatch) -> None: - timestamps = {Chain.MAINNET: NOW - 5 * HOUR, Chain.BASE: NOW - 120} - monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: timestamps[chain]) + 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 [c.chain for c in result] == [Chain.MAINNET, Chain.BASE] + 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 @@ -59,9 +91,10 @@ def test_collect_freshness_computes_lag_and_sorts_by_chain(monkeypatch: pytest.M def test_collect_freshness_skips_chains_without_a_processed_block(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: NOW) - rows = [{"chain_id": 1, "latest_processed_block": 0, "block_height": 0}] - assert freshness.collect_freshness(rows, 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: @@ -69,28 +102,46 @@ def test_collect_freshness_marks_lag_unknown_when_rpc_fails(monkeypatch: pytest. result = freshness.collect_freshness(_rows(), NOW) - assert all(chain.lag_seconds is None for chain in result) + 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(chain.is_stale(HOUR) for chain in result) + assert not any(entry.is_stale(HOUR) for entry in result) def test_is_stale_uses_threshold() -> None: - chain = ChainFreshness(chain=Chain.MAINNET, latest_processed_block=100, lag_seconds=HOUR + 1) + 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 chain.is_stale(HOUR) - assert not chain.is_stale(2 * HOUR) + assert freshness.missing_chains(freshness.collect_freshness(_rows(), NOW)) == [] -def test_build_stale_message_lists_every_lagging_chain() -> None: +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_stale_message(stale, HOUR) + 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 @@ -135,12 +186,22 @@ def test_fetch_chain_metadata_raises_when_url_missing(monkeypatch: pytest.Monkey 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, NOW, 6 * HOUR) == stale + assert freshness.chains_to_alert(stale, missing, NOW, 6 * HOUR) == (stale, missing) - freshness._set_last_alert_timestamp(1, NOW) - assert freshness.chains_to_alert(stale, NOW + HOUR, 6 * HOUR) == [] - assert freshness.chains_to_alert(stale, NOW + 6 * HOUR, 6 * HOUR) == stale + 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( @@ -151,8 +212,8 @@ def test_main_alerts_once_per_cooldown_then_reports_recovery( ) monkeypatch.setattr("sys.argv", ["check_indexer_freshness.py"]) - stale_timestamps = {Chain.MAINNET: NOW - 5 * HOUR, Chain.BASE: NOW - 120} - monkeypatch.setattr(freshness, "fetch_block_timestamp", lambda chain, block: stale_timestamps[chain]) + 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() @@ -165,7 +226,7 @@ def test_main_alerts_once_per_cooldown_then_reports_recovery( assert len(sent) == 1 # Once mainnet catches up, the recovery note fires and clears the state. - stale_timestamps[Chain.MAINNET] = NOW - 60 + timestamps[Chain.MAINNET] = NOW - 60 freshness.main() assert len(sent) == 2 assert "caught up" in sent[1] @@ -174,6 +235,40 @@ def test_main_alerts_once_per_cooldown_then_reports_recovery( 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: