diff --git a/.env.example b/.env.example index d20fb74c..830c4460 100644 --- a/.env.example +++ b/.env.example @@ -22,6 +22,7 @@ ENVIO_GRAPHQL_URL="https://indexer.hyperindex.xyz/3fec0a4/v1/graphql" # Event subscriber pilot (protocols/timelock/event_subscriber.py, local output only) # EVENT_SUBSCRIBER_STREAMS=timelock # comma-separated: timelock, morpho (needs MorphoGovernanceEvent in Envio) +# MORPHO_LIVE_DELIVERY=shadow # shadow (errors chat) | observe | production (Morpho channel) # ENVIO_GRAPHQL_WS_URL=wss://... # default: ENVIO_GRAPHQL_URL with https → wss # TIMELOCK_SUBSCRIBER_TRANSPORT=both # both | ws | http # TIMELOCK_SUBSCRIBER_NOTIFY=none # errors: post pilot notices to the errors chat diff --git a/deploy/runbook.md b/deploy/runbook.md index c616e29f..32bf101d 100644 --- a/deploy/runbook.md +++ b/deploy/runbook.md @@ -290,6 +290,21 @@ at each chain's indexed head; rerun it to confirm the range queries succeed. To disable the stream, drop it from `EVENT_SUBSCRIBER_STREAMS` and restart; its inbox is kept. +The stream's live consumer starts in `shadow` mode: for each new event it posts what +the daily governance jobs would alert to the internal errors chat as +`[morpho-live-shadow]` (`report` shows the summary), for comparison with the daily +alerts. Without an errors chat it only records them (`observe`, also selectable with +`MORPHO_LIVE_DELIVERY=observe`). The daily jobs keep alerting the Morpho group in both +modes. Events from before the consumer's first run are recorded as history and never +alerted. + +**Cutover to production** ([#450](https://github.com/yearn/monitoring/issues/450)): once a +shadow period matched the daily Morpho alerts, set `MORPHO_LIVE_DELIVERY=production` +and restart. Live alerts then go to the Morpho channel within seconds of the event. +The daily jobs keep running: both paths take the same per-vault lock in +`/srv/cache` and share the cache that records what was alerted, so each change is +alerted once, by whichever path sees it first. To roll back, set `shadow` and restart. + Supervision: the unit is `Type=notify` with `WatchdogSec=120`. The heartbeat pings the watchdog only while HTTP reconciliation keeps completing, so a wedged worker is killed and restarted. `Restart=always` also covers a crash, after which @@ -317,6 +332,7 @@ Set in `/etc/monitoring/.env`, then restart the unit: | Variable | Default | Meaning | |---|---|---| | `EVENT_SUBSCRIBER_STREAMS` | `timelock` | Streams to follow: `timelock`, `morpho`, comma-separated | +| `MORPHO_LIVE_DELIVERY` | `shadow` | Live Morpho governance decisions: `shadow` posts them to the internal errors chat (records only without one); `observe` records them only; `production` sends them to the Morpho channel | | `TIMELOCK_SUBSCRIBER_TRANSPORT` | `both` | `both`, `ws`, or `http` (every stream) | | `TIMELOCK_SUBSCRIBER_NOTIFY` | `none` | `errors` posts pilot notices to the internal errors chat (needs `TELEGRAM_TOPIC_ID_ERRORS` or `TELEGRAM_CHAT_ID_ERRORS`) | | `TIMELOCK_SUBSCRIBER_RECONCILE_SECONDS` | `60` | HTTP reconcile interval | diff --git a/protocols/morpho/README.md b/protocols/morpho/README.md index e46be988..c6bbd2ae 100644 --- a/protocols/morpho/README.md +++ b/protocols/morpho/README.md @@ -203,9 +203,34 @@ Both scripts share `MORPHO_FILENAME` (default `cache-id.txt`). New key types add [`governance_stream.py`](./governance_stream.py) defines the `morpho` stream of the continuous event subscriber ([#447](https://github.com/yearn/monitoring/issues/447), part of [#437](https://github.com/yearn/monitoring/issues/437)). It follows `MorphoGovernanceEvent`, which [yearn-envio#71](https://github.com/yearn/yearn-envio/pull/71) indexes for every V1 and V2 vault in [`config.py`](./config.py): V1 `SubmitCap`, `SubmitMarketRemoval`, `SubmitTimelock`; V2 `Submit`, `Accept`, `SetOwner`, `SetCurator`, `AddAdapter`, `RemoveAdapter`. Adapter-internal timelocks (`burnShares`) are not indexed. -It runs in the timelock subscriber's process with `EVENT_SUBSCRIBER_STREAMS=timelock,morpho` and uses the same WebSocket + HTTP reconciliation, per-chain checkpoints, reorg tracking, health, and report (see the [timelock README](../timelock/README.md#continuous-subscriber-pilot)); its inbox is `$CACHE_DIR/morpho_subscriber.db`. It is observation only: **it sends no alerts**, and the daily `governance.py` / `governance_v2.py` jobs remain the alerting path. With notices enabled, each new event is posted to the internal errors chat as `[morpho-subscriber]` with the vault, its generation, the sender, and the ABI-named arguments. Lifecycle rules and grouped delivery are [#448](https://github.com/yearn/monitoring/issues/448); comparing with the daily snapshots is [#449](https://github.com/yearn/monitoring/issues/449). +It runs in the timelock subscriber's process with `EVENT_SUBSCRIBER_STREAMS=timelock,morpho` and uses the same WebSocket + HTTP reconciliation, per-chain checkpoints, reorg tracking, health, and report (see the [timelock README](../timelock/README.md#continuous-subscriber-pilot)); its inbox is `$CACHE_DIR/morpho_subscriber.db`. It sends no alerts to the Morpho group: the daily `governance.py` / `governance_v2.py` jobs remain the alerting path (see live decisions below). With notices enabled, each new event is posted to the internal errors chat as `[morpho-subscriber]` with the vault, its generation, the sender, and the ABI-named arguments. Comparing with the daily snapshots is [#449](https://github.com/yearn/monitoring/issues/449). -Enable it only once the Envio deployment serves `MorphoGovernanceEvent` and has backfilled all five chains; until then its reconcile cycles and WebSocket fail with `field 'MorphoGovernanceEvent' not found` (without affecting the timelock stream). Its first reconcile starts at each chain's indexed head, so history is never announced. +#### Live decisions ([#448](https://github.com/yearn/monitoring/issues/448)) + +[`live_governance.py`](./live_governance.py) runs inside the `morpho` stream's worker and decides, for every newly indexed event, what the daily jobs would alert. The indexer has no V1 revoke/accept events and no V2 `Revoke`, so an event is only a trigger: [`live_rules.py`](./live_rules.py) reads the vault's current state on-chain, at a block at or after the event (an RPC node behind the event is retried), and runs the daily jobs' own checks on it: + +| Event | Check (same function as the daily job) | +|---|---| +| V1 `SubmitCap`, `SubmitMarketRemoval` | that market's pending cap and removal checks, including the potential-loss alert | +| V1 `SubmitTimelock` | the timelock (and guardian) check | +| V2 `Submit` | `executableAt(data)` must still be set, then the pending-operation diff (loss alerts, grouping of new submissions) | +| V2 `SetOwner`, `SetCurator` | the owner/curator diff | +| V2 `AddAdapter`, `RemoveAdapter` | the adapter set diff | +| V2 `Accept` | none: executed/revoked notes stay with the daily job | + +So content, severity, per-vault grouping, and dedup match the daily jobs, and an operation revoked or executed before the read is no longer pending and is not alerted. The checks read the daily jobs' cache (an operation they already alerted is not alerted again) but write it only in `production` mode, so in the other modes the daily jobs keep alerting exactly as before. That cache is the daily profile's namespace (`MORPHO_FILENAME` from `automation/jobs.yaml`, applied by [`daily_governance_cache.py`](./daily_governance_cache.py) before `utils.cache` loads); the subscriber refuses to start the Morpho consumer if it resolved another one. An adapter whose `realAssets()` cannot be read fails the evaluation and is retried, never treated as empty. + +Decisions and alert requests are stored durably next to the inbox (`morpho_live_decisions`, `morpho_live_alerts` in `morpho_subscriber.db`, see [`live_store.py`](./live_store.py)), one decision per event revision: a restart resumes undecided events, and a reorg that rewrites an event re-evaluates it. Events from blocks before the consumer's first run on the inbox (`live_baseline_at`) are recorded as `historical` and never evaluated. A failing vault retries with capped backoff, and is left out of the batch meanwhile so it cannot crowd out other vaults. + +`MORPHO_LIVE_DELIVERY` picks the delivery mode: + +- `shadow` (default): each request is posted to the internal errors chat as `[morpho-live-shadow]`, as plain text split to Telegram's limit, in order, retrying a failed or skipped send (e.g. a missing bot token) until Telegram accepts it. A request whose source events a reorg removed or rewrote before delivery is `invalidated` instead. Shadow requires the errors chat (`TELEGRAM_TOPIC_ID_ERRORS` or `TELEGRAM_CHAT_ID_ERRORS`) and falls back to `observe` without it, so its output never reaches the Morpho group. +- `observe`: requests are recorded as `observed`; nothing is sent. +- `production`: alerts go to the Morpho channel, like the daily jobs. Per vault, under a `governance_lock` that the daily `governance.py` / `governance_v2.py` jobs also take, the consumer evaluates, sends every message, and only then commits the daily jobs' cache writes. So each change is alerted once, by whichever path sees it first, and the daily jobs keep catching anything the live path missed (indexer gaps, outages) plus resolutions and guardian changes. A failed send, or one Telegram skips for missing credentials, commits nothing and leaves the events undecided; they are re-evaluated and re-sent on retry (a duplicate beats a missed alert, as in the daily jobs). + +Switch to `production` only after a shadow period matched the daily alerts ([#450](https://github.com/yearn/monitoring/issues/450)). The lock is a file in `CACHE_DIR` (`morpho_governance_.lock`), so both processes must share it, as the `monitoring` and `monitoring-timelock-subscriber` units do. `event_subscriber.py --streams morpho report` adds a live-decision summary (outcomes, block-to-decision delay, request statuses); `health` also fails when events stay undecided or shadow alerts stay undelivered for more than 10 minutes. + +Enable the stream only once the Envio deployment serves `MorphoGovernanceEvent` and has backfilled all five chains; until then its reconcile cycles and WebSocket fail with `field 'MorphoGovernanceEvent' not found` (without affecting the timelock stream). Its first reconcile starts at each chain's indexed head, so history is never announced. ### Limitations diff --git a/protocols/morpho/_alerts.py b/protocols/morpho/_alerts.py index 361375f8..dac945e3 100644 --- a/protocols/morpho/_alerts.py +++ b/protocols/morpho/_alerts.py @@ -15,12 +15,14 @@ from __future__ import annotations +from contextlib import AbstractContextManager from dataclasses import dataclass, field from datetime import datetime from functools import partial from typing import Any, Callable, List from utils.alert import Alert, AlertSeverity, send_alert +from utils.file_lock import file_lock from utils.telegram import MAX_MESSAGE_LENGTH # Ascending severity — a grouped alert is sent at the highest of its sections. @@ -44,6 +46,16 @@ REMINDER_WINDOW_SECONDS = 24 * 60 * 60 +def governance_lock(vault_address: str) -> AbstractContextManager[None]: + """Lock one vault's governance check → send → cache-write sequence across processes. + + The daily jobs and the live consumer's production mode read and write the + same cache keys; holding this lock around the whole sequence keeps them from + both treating one change as new and alerting it twice. + """ + return file_lock(f"morpho_governance_{vault_address.lower()}") + + @dataclass(frozen=True) class LossNotice: """A potential-loss alert that is due for one pending operation.""" @@ -155,30 +167,42 @@ def split_into_messages(alerts: List[VaultAlert], budget: int) -> List[List[str] return parts +def render_vault_alerts(header: str, alerts: List[VaultAlert]) -> List[tuple[AlertSeverity, str]]: + """Render buffered sections into the ``(severity, message)`` pairs ``send_vault_alerts`` sends. + + Standalone (potential-loss) sections come first, one message each, under a + ``LOSS_PREFIX`` header; the remaining sections are grouped under ``header`` at + their highest severity and split into "(i/N)" parts if long. + """ + messages: List[tuple[AlertSeverity, str]] = [] + for alert in alerts: + if alert.standalone: + messages += _render_group(f"{LOSS_PREFIX} — {header}", [alert]) + messages += _render_group(header, [a for a in alerts if not a.standalone]) + return messages + + def send_vault_alerts(header: str, alerts: List[VaultAlert], protocol: str) -> None: """Send the buffered sections as one Telegram message, or "(i/N)" parts if long. No-op when ``alerts`` is empty so callers don't have to guard. Every part carries the same header and the highest severity of the whole group, so a - LOW section bundled with an owner change still pings the channel. - - Standalone (potential-loss) sections go out first, one message each, under a - ``LOSS_PREFIX`` header; the remaining sections are grouped as usual. + LOW section bundled with an owner change still pings the channel. See + ``render_vault_alerts`` for the message layout. """ - for alert in alerts: - if alert.standalone: - _send_group(f"{LOSS_PREFIX} — {header}", [alert], protocol) - _send_group(header, [a for a in alerts if not a.standalone], protocol) + for severity, message in render_vault_alerts(header, alerts): + send_alert(Alert(severity, message, protocol)) -def _send_group(header: str, alerts: List[VaultAlert], protocol: str) -> None: - """Send sections under one header at their highest severity, split into parts if long.""" +def _render_group(header: str, alerts: List[VaultAlert]) -> List[tuple[AlertSeverity, str]]: + """Render sections under one header at their highest severity, split into parts if long.""" if not alerts: - return + return [] severity = max((a.severity for a in alerts), key=_SEVERITY_ORDER.index) parts = split_into_messages(alerts, MAX_MESSAGE_LENGTH - _MESSAGE_OVERHEAD - len(header)) total = len(parts) + messages = [] for index, bodies in enumerate(parts, start=1): suffix = f" ({index}/{total})" if total > 1 else "" - message = f"{header}{suffix}\n\n" + SECTION_SEPARATOR.join(bodies) - send_alert(Alert(severity, message, protocol)) + messages.append((severity, f"{header}{suffix}\n\n" + SECTION_SEPARATOR.join(bodies))) + return messages diff --git a/protocols/morpho/daily_governance_cache.py b/protocols/morpho/daily_governance_cache.py new file mode 100644 index 00000000..b2b759fd --- /dev/null +++ b/protocols/morpho/daily_governance_cache.py @@ -0,0 +1,52 @@ +"""Point this process's Morpho cache at the daily governance jobs' namespace. + +The daily profile in ``automation/jobs.yaml`` runs ``governance.py`` and +``governance_v2.py`` with ``MORPHO_FILENAME`` set to its own namespace, and +``utils.cache`` resolves that name once, at import. The continuous subscriber +runs outside that profile, so without this its live governance checks would read +(and in production, write) a different namespace: no role/set baselines, no +record of what the daily jobs already alerted. + +Importing this module sets ``MORPHO_FILENAME`` from the daily profile, unless the +process already sets it explicitly, so it must be imported before anything +imports ``utils.cache``. ``check_daily_governance_cache`` verifies at startup that +the order held. +""" + +import os + +from automation.config import load_jobs_config + +DAILY_PROFILE = "daily" + + +def daily_morpho_filename() -> str: + """Return the ``MORPHO_FILENAME`` the daily governance jobs run with.""" + filename = load_jobs_config().profiles[DAILY_PROFILE].env.get("MORPHO_FILENAME") + if not filename: + raise RuntimeError(f"automation/jobs.yaml profile {DAILY_PROFILE!r} sets no MORPHO_FILENAME") + return filename + + +def use_daily_governance_cache() -> str: + """Set ``MORPHO_FILENAME`` to the daily jobs' namespace unless already set, and return it.""" + return os.environ.setdefault("MORPHO_FILENAME", daily_morpho_filename()) + + +def check_daily_governance_cache() -> None: + """Fail unless ``utils.cache`` resolved the Morpho cache to the daily jobs' namespace. + + Raises: + RuntimeError: ``utils.cache`` was imported before this module, or with another name. + """ + from utils import cache # noqa: PLC0415 - checks the binding the import above must not have fixed early + + expected = os.environ.get("MORPHO_FILENAME") or daily_morpho_filename() + if os.path.basename(cache.morpho_filename) != expected: + raise RuntimeError( + f"utils.cache uses {cache.morpho_filename}, not the daily governance namespace {expected}: " + f"import {__name__} before anything imports utils.cache" + ) + + +use_daily_governance_cache() diff --git a/protocols/morpho/governance.py b/protocols/morpho/governance.py index b40be198..9b132090 100644 --- a/protocols/morpho/governance.py +++ b/protocols/morpho/governance.py @@ -4,7 +4,7 @@ from web3 import Web3 -from protocols.morpho._alerts import VaultDiff, due_loss_notice, send_vault_alerts +from protocols.morpho._alerts import VaultDiff, due_loss_notice, governance_lock, send_vault_alerts from protocols.morpho._shared import ( MORPHO_BLUE_ABI_PATH, PROTOCOL, @@ -159,7 +159,7 @@ def _load_vault_market_ids( return list(set(queued_markets) | pending_markets) -def _load_market_governance_states( +def load_market_governance_states( morpho_contract: Any, market_ids: list[bytes], client: Any, @@ -219,7 +219,7 @@ def _try_total_assets(morpho_contract: Any, block_number: int) -> int | None: return None -def _vault_header(name: str, vault_address: str, chain: Chain) -> str: +def vault_header(name: str, vault_address: str, chain: Chain) -> str: """One-line header for the vault's grouped alert.""" return f"[{name}]({get_vault_url(vault_address, chain)}) on {chain.name}" @@ -359,7 +359,7 @@ def check_markets_pending_cap( ) -> None: """Check V1 market cap and removal governance for one vault.""" market_ids = _load_vault_market_ids(morpho_contract, chain, client, block_number) - for state in _load_market_governance_states(morpho_contract, market_ids, client, block_number): + for state in load_market_governance_states(morpho_contract, market_ids, client, block_number): check_market_governance_state(name, state, chain, diff) @@ -422,19 +422,21 @@ def get_data_for_chain(chain: Chain) -> None: for vault in vaults: morpho_contract = client.eth.contract(address=vault.address, abi=ABI_MORPHO) # Buffer every finding for this vault, then send them as one message and - # only then record them as alerted. - diff = VaultDiff() - check_markets_pending_cap( - vault.name, - morpho_contract, - chain, - client, - diff, - block_number=block_number, - ) - check_timelock_and_guardian(morpho_contract, client, diff, block_number) - send_vault_alerts(_vault_header(vault.name, vault.address, chain), diff.alerts, PROTOCOL) - diff.commit() + # only then record them as alerted. The lock keeps the live consumer from + # alerting the same change in between. + with governance_lock(vault.address): + diff = VaultDiff() + check_markets_pending_cap( + vault.name, + morpho_contract, + chain, + client, + diff, + block_number=block_number, + ) + check_timelock_and_guardian(morpho_contract, client, diff, block_number) + send_vault_alerts(vault_header(vault.name, vault.address, chain), diff.alerts, PROTOCOL) + diff.commit() def main() -> None: diff --git a/protocols/morpho/governance_v2.py b/protocols/morpho/governance_v2.py index 8f6582e6..fed1265b 100644 --- a/protocols/morpho/governance_v2.py +++ b/protocols/morpho/governance_v2.py @@ -32,7 +32,7 @@ from web3 import Web3 from protocols.morpho import adapter_timelocks_v2 -from protocols.morpho._alerts import VaultDiff, due_loss_notice, send_vault_alerts +from protocols.morpho._alerts import VaultDiff, due_loss_notice, governance_lock, send_vault_alerts from protocols.morpho._shared import ( PROTOCOL, MorphoV2MonitoringError, @@ -304,7 +304,7 @@ def _pending_function_key(snapshot: V2GovernanceSnapshot, data_hash: str) -> str return str(morpho_key(snapshot.address.lower(), data_hash, PENDING_FUNCTION_TYPE)) -def _vault_header(snapshot: V2GovernanceSnapshot) -> str: +def vault_header(snapshot: V2GovernanceSnapshot) -> str: """One-line header for the grouped alert: ``V2 [name](url) on chain``.""" return f"V2 [{snapshot.name}]({get_vault_url(snapshot.address, snapshot.chain)}) on {snapshot.chain.name}" @@ -379,7 +379,7 @@ def _check_pending_loss( diff.defer(_write, reminder_key, pc.valid_at) -def _alert_pending_new( +def alert_pending_new( snapshot: V2GovernanceSnapshot, pending: List[tuple[PendingConfig, str]], diff: VaultDiff, @@ -464,31 +464,50 @@ def _alert_set_diff(set_name: str, added: set[str], removed: set[str], diff: Vau # ---------------------------------------------------------------------------- +def diff_pending_config( + snapshot: V2GovernanceSnapshot, pc: PendingConfig, diff: VaultDiff +) -> tuple[PendingConfig, str] | None: + """Diff one pending operation against the cache. + + A potential loss is buffered on ``diff`` directly (as its own message). Any + other operation not alerted yet is returned as ``(pc, label)`` for the caller + to group with the vault's other new submissions via ``alert_pending_new``. + Shared by the daily snapshot diff and the live event consumer, so both alert + the same operations with the same content and severity. + """ + operation_label = _operation_label(snapshot, pc) + diff.defer(_write, _pending_function_key(snapshot, pc.data_hash), _operation_function_name(pc, operation_label)) + cache_key = morpho_key(snapshot.address.lower(), pc.data_hash, PENDING_TYPE) + last = _read_int(cache_key) + if last == EXECUTED: + return None + new = None + # A loss is tracked on its own key, so an operation first seen with nothing + # at risk still escalates once funds arrive. + loss = pending_loss(snapshot, pc) + if loss is not None: + _check_pending_loss(snapshot, pc, operation_label, loss, diff) + elif last != pc.valid_at: + new = (pc, operation_label) + if last != pc.valid_at: + diff.defer(_write, cache_key, pc.valid_at) + return new + + def _diff_pending(snapshot: V2GovernanceSnapshot, diff: VaultDiff) -> None: + """Diff every pending operation and detect the ones that left ``pendingConfigs``.""" addr = snapshot.address.lower() current_keys: set[str] = set() new_pending: List[tuple[PendingConfig, str]] = [] for pc in snapshot.pending_configs: current_keys.add(pc.data_hash) - operation_label = _operation_label(snapshot, pc) - diff.defer(_write, _pending_function_key(snapshot, pc.data_hash), _operation_function_name(pc, operation_label)) - cache_key = morpho_key(addr, pc.data_hash, PENDING_TYPE) - last = _read_int(cache_key) - if last == EXECUTED: - continue - # A loss is tracked on its own key, so an operation first seen with nothing - # at risk still escalates once funds arrive. - loss = pending_loss(snapshot, pc) - if loss is not None: - _check_pending_loss(snapshot, pc, operation_label, loss, diff) - elif last != pc.valid_at: - new_pending.append((pc, operation_label)) - if last != pc.valid_at: - diff.defer(_write, cache_key, pc.valid_at) + new = diff_pending_config(snapshot, pc, diff) + if new is not None: + new_pending.append(new) # Group all other newly-submitted operations for this vault into one section. - _alert_pending_new(snapshot, new_pending, diff) + alert_pending_new(snapshot, new_pending, diff) # Detect resolved entries: anything in last-run's index that isn't in the # current pending list. Reading a function-name key here is safe despite the @@ -512,7 +531,8 @@ def _diff_pending(snapshot: V2GovernanceSnapshot, diff: VaultDiff) -> None: diff.defer(_write, index_key, ",".join(sorted(current_keys))) -def _diff_single_role(snapshot: V2GovernanceSnapshot, role: str, current: str, diff: VaultDiff) -> None: +def diff_single_role(snapshot: V2GovernanceSnapshot, role: str, current: str, diff: VaultDiff) -> None: + """Alert HIGH when ``role`` (owner/curator) differs from the cached holder; the first run seeds silently.""" cache_key = morpho_key(snapshot.address.lower(), role, ROLE_TYPE) last = _read_str(cache_key) cur_lc = current.lower() @@ -521,7 +541,8 @@ def _diff_single_role(snapshot: V2GovernanceSnapshot, role: str, current: str, d diff.defer(_write, cache_key, cur_lc) -def _diff_set(snapshot: V2GovernanceSnapshot, set_name: str, current: List[str], diff: VaultDiff) -> None: +def diff_set(snapshot: V2GovernanceSnapshot, set_name: str, current: List[str], diff: VaultDiff) -> None: + """Alert LOW on members added to or removed from ``set_name`` since the cached set; the first run seeds silently.""" cache_key = morpho_key(snapshot.address.lower(), set_name, SET_TYPE) last_str = _read_str(cache_key) last_set = {a for a in last_str.split(",") if a} if last_str else set() @@ -556,17 +577,19 @@ def diff_and_alert( ``adapter_states`` are the on-chain adapter timelock reads; ``None`` skips the adapter diff (e.g. when the RPC read failed) so its cache is left untouched. """ - diff = VaultDiff() - _diff_pending(snapshot, diff) - _diff_single_role(snapshot, "owner", snapshot.owner, diff) - _diff_single_role(snapshot, "curator", snapshot.curator, diff) - _diff_set(snapshot, "sentinels", snapshot.sentinels, diff) - _diff_set(snapshot, "allocators", snapshot.allocators, diff) - _diff_set(snapshot, "adapters", snapshot.adapters, diff) - for state in adapter_states or []: - adapter_timelocks_v2.diff_adapter(snapshot.chain, snapshot.total_assets, state, diff) - send_vault_alerts(_vault_header(snapshot), diff.alerts, PROTOCOL) - diff.commit() + # The lock keeps the live consumer from alerting the same change in between. + with governance_lock(snapshot.address): + diff = VaultDiff() + _diff_pending(snapshot, diff) + diff_single_role(snapshot, "owner", snapshot.owner, diff) + diff_single_role(snapshot, "curator", snapshot.curator, diff) + diff_set(snapshot, "sentinels", snapshot.sentinels, diff) + diff_set(snapshot, "allocators", snapshot.allocators, diff) + diff_set(snapshot, "adapters", snapshot.adapters, diff) + for state in adapter_states or []: + adapter_timelocks_v2.diff_adapter(snapshot.chain, snapshot.total_assets, state, diff) + send_vault_alerts(vault_header(snapshot), diff.alerts, PROTOCOL) + diff.commit() # ---------------------------------------------------------------------------- diff --git a/protocols/morpho/live_governance.py b/protocols/morpho/live_governance.py new file mode 100644 index 00000000..b28e5cc3 --- /dev/null +++ b/protocols/morpho/live_governance.py @@ -0,0 +1,379 @@ +"""Live Morpho governance consumer: indexed events → daily-job decisions → recorded or shadow alerts. + +Runs inside the ``morpho`` stream's worker (#448). Ingestion only writes the +inbox; this consumer reads it on its own schedule (woken after each ingest, with +a polling fallback), so a slow RPC read or Telegram call never delays intake. + +For every inbox event without a decision for its current revision, in chain +order and grouped per vault, it asks ``live_rules`` what the daily jobs would +alert, renders that with the daily jobs' own grouping (``render_vault_alerts``), +and stores the decision and the alert requests durably (``live_store``): + +- **Historical replay never alerts.** Events whose block predates the consumer's + baseline (the first time it ran on this inbox) are recorded as ``historical``. + Events from later blocks are evaluated even when first seen after a restart. +- **Failures retry.** An RPC node behind the event's block, or any read error, + leaves the vault's events undecided and retries with capped backoff. +- **Delivery mode** (``MORPHO_LIVE_DELIVERY``, ``shadow`` by default in the + subscriber CLI; ``LiveConfig`` itself defaults to ``observe``): ``observe`` + records requests as ``observed`` and sends nothing. ``shadow`` posts each + request to the internal errors chat, labelled ``[morpho-live-shadow]``, and only + when that chat is configured, so shadow output never reaches the Morpho group. + A failed or skipped send stays pending and is retried; a request whose source + events a reorg removed or rewrote is invalidated instead. +- ``production`` sends to the Morpho channel exactly like the daily jobs: under + the vault's ``governance_lock`` (which the daily jobs take too) it evaluates, + sends every message, and only then commits the daily jobs' cache writes, so + each change is alerted once by whichever path sees it first. A failed send + commits nothing and leaves the events undecided: they are re-evaluated and + re-sent on retry (a duplicate beats a missed alert, as in the daily jobs). + Enable it only after shadow mode matched the daily alerts (#450). +""" + +import asyncio +import time +from collections.abc import Callable +from dataclasses import dataclass + +from protocols.morpho._alerts import governance_lock, render_vault_alerts, split_body +from protocols.morpho._shared import PROTOCOL +from protocols.morpho.config import VaultConfig +from protocols.morpho.live_rules import ( + REASON_ALERTED, + GovernanceEvent, + RpcBehindError, + VaultEvaluation, + evaluate_v1, + evaluate_v2, + vault_config, +) +from protocols.morpho.live_store import ( + OUTCOME_ALERTED, + OUTCOME_HISTORICAL, + OUTCOME_NO_ALERT, + STATUS_OBSERVED, + STATUS_PENDING, + STATUS_SENT, + AlertRequest, + Decision, + LiveStore, + PendingEvent, +) +from utils.alert import Alert, AlertSeverity, send_alert +from utils.chains import Chain +from utils.logger import get_logger +from utils.subscriber.worker import backoff_delay +from utils.telegram import ERROR_CHANNEL, MAX_MESSAGE_LENGTH, error_channel_configured, send_telegram_message +from utils.web3_wrapper import ChainManager + +logger = get_logger("morpho.live_governance") + +MODE_OBSERVE = "observe" +MODE_SHADOW = "shadow" +MODE_PRODUCTION = "production" +DELIVERY_MODES = (MODE_OBSERVE, MODE_SHADOW, MODE_PRODUCTION) + +# Inbox status key: when the consumer first ran on this inbox. Older blocks are history. +BASELINE_STATUS_KEY = "live_baseline_at" +SHADOW_LABEL = "morpho-live-shadow" +# Alert-history key: internal-only, so shadow messages stay off the public Morpho page. +SHADOW_ALERT_PROTOCOL = "yearn-internal" + +Evaluator = Callable[[Chain, VaultConfig, int, list[GovernanceEvent]], VaultEvaluation] +SendFn = Callable[[AlertRequest], None] +PublishFn = Callable[[AlertSeverity, str], None] + + +class ShadowChannelMissing(Exception): + """Shadow delivery needs the internal errors chat; it never falls back to a protocol group.""" + + +class NotDelivered(Exception): + """Telegram skipped the message (missing credentials or debug mode); it must not count as sent.""" + + +# Event ids listed in a shadow message header; the rest are counted. +_SHADOW_MAX_IDS = 5 + + +def evaluate_on_chain(chain: Chain, vault: VaultConfig, version: int, events: list[GovernanceEvent]) -> VaultEvaluation: + """Evaluate one vault's events against its current on-chain state.""" + client = ChainManager.get_client(chain) + return (evaluate_v1 if version == 1 else evaluate_v2)(client, chain, vault, events) + + +def shadow_texts(request: AlertRequest) -> list[str]: + """Return the exact plain-text messages shadow delivery posts for ``request``. + + The header lists at most a few event ids, and the whole text is split by line + to Telegram's limit, so nothing is truncated however many events contributed. + """ + ids = request.event_ids + shown = ", ".join(ids[:_SHADOW_MAX_IDS]) + ( + f" (+{len(ids) - _SHADOW_MAX_IDS} more)" if len(ids) > _SHADOW_MAX_IDS else "" + ) + text = f"[{SHADOW_LABEL}] shadow → morpho, {request.severity}, events {shown}\n\n{request.message}" + return split_body(text, MAX_MESSAGE_LENGTH) + + +def send_shadow(request: AlertRequest) -> None: + """Post one would-be Morpho alert to the internal errors chat, as plain text. + + Sent straight to the errors channel (no fallback routing, no escaping), and + only counted as sent when Telegram accepted every part. + + Raises: + ShadowChannelMissing: No errors chat is configured. + NotDelivered: Telegram skipped a part (e.g. the bot token is missing). + TelegramError: A send failed. + """ + if not error_channel_configured(): + raise ShadowChannelMissing("no errors chat configured for shadow delivery") + for text in shadow_texts(request): + delivered = send_telegram_message( + text, + ERROR_CHANNEL, + True, + plain_text=True, + source="morpho_live_shadow", + origin_protocol=SHADOW_ALERT_PROTOCOL, + channel=ERROR_CHANNEL, + ) + if not delivered: + raise NotDelivered("errors chat skipped the shadow message (missing Telegram credentials?)") + + +def publish_to_morpho(severity: AlertSeverity, message: str) -> None: + """Send one rendered message to the Morpho channel, as ``send_vault_alerts`` does. + + Raises: + TelegramError: The send failed; the caller must not commit its cache writes. + NotDelivered: Telegram skipped it (e.g. missing Morpho credentials); nor then. + """ + if not send_alert(Alert(severity, message, PROTOCOL)): + raise NotDelivered("Morpho channel skipped the alert (missing Telegram credentials?)") + + +@dataclass(frozen=True) +class LiveConfig: + """Mode and timing of the consumer loops.""" + + mode: str = MODE_OBSERVE + # Fallback poll; ingestion wakes the consumer as soon as rows land. + poll_interval: float = 5.0 + retry_initial: float = 5.0 + retry_max: float = 300.0 + delivery_interval: float = 1.0 + batch_limit: int = 500 + + +class LiveGovernanceConsumer: + """Evaluates new Morpho governance events and records (or shadow-sends) the alerts they imply.""" + + def __init__( + self, + store: LiveStore, + config: LiveConfig = LiveConfig(), + evaluate: Evaluator = evaluate_on_chain, + send: SendFn = send_shadow, + now: Callable[[], float] = time.time, + publish: PublishFn = publish_to_morpho, + ) -> None: + if config.mode not in DELIVERY_MODES: + raise ValueError(f"delivery mode must be one of {DELIVERY_MODES}, got {config.mode!r}") + self.store = store + self.config = config + self.evaluate = evaluate + self.send = send + self.publish = publish + self.now = now + self.baseline: float | None = None + self._wake = asyncio.Event() + # (chain id, vault) -> (consecutive failures, retry not before) + self._retry: dict[tuple[int, str], tuple[int, float]] = {} + self._delivery_failures = 0 + + def wake(self) -> None: + """Ask the consumer to look for new events now (called after each ingest).""" + self._wake.set() + + def load_baseline(self) -> float: + """Load the baseline, recording it the first time the consumer runs on this inbox.""" + if self.baseline is None: + stored = self.store.inbox.status().get(BASELINE_STATUS_KEY, (None, 0.0))[0] + if stored is None: + stored = self.now() + self.store.inbox.set_status(BASELINE_STATUS_KEY, stored, stored) + self.baseline = float(stored) + return self.baseline + + async def run(self) -> None: + """Run the evaluation loop, and the delivery loop in shadow mode, until cancelled.""" + await asyncio.to_thread(self.load_baseline) + loops = [self._evaluation_loop()] + if self.config.mode == MODE_SHADOW: + loops.append(self._delivery_loop()) + await asyncio.gather(*loops) + + async def _evaluation_loop(self) -> None: + """Evaluate pending events whenever woken, or every ``poll_interval``.""" + while True: + self._wake.clear() + try: + await asyncio.to_thread(self.process_once) + except Exception: # noqa: BLE001 - a store or rule bug must not end the consumer + logger.exception("live governance evaluation pass failed") + try: + await asyncio.wait_for(self._wake.wait(), timeout=self.config.poll_interval) + except TimeoutError: + pass + + def process_once(self) -> int: + """Evaluate every pending event once, per vault, in chain order. + + Returns: + The number of events decided in this pass. + """ + baseline = self.load_baseline() + # Backed-off vaults are left out before the batch limit applies, so a failing + # vault's backlog can never crowd out other vaults' new events. + deferred = {key for key, (_, not_before) in self._retry.items() if self.now() < not_before} + groups: dict[tuple[int, str], list[PendingEvent]] = {} + for event in self.store.undecided(self.config.batch_limit, exclude=deferred): + groups.setdefault((event.chain_id, event.vault_address), []).append(event) + decided = 0 + for key, events in groups.items(): + failures, not_before = self._retry.get(key, (0, 0.0)) + if self.now() < not_before: + continue + # Block timestamps are whole seconds; a block in the baseline's own second counts. + historical = [e for e in events if e.block_timestamp < int(baseline)] + fresh = [e for e in events if e.block_timestamp >= int(baseline)] + if historical: + self._record(historical, OUTCOME_HISTORICAL, "block before the live baseline") + decided += len(historical) + if not fresh: + continue + try: + decided += self._evaluate_vault(fresh) + except RpcBehindError as exc: + logger.info("%s on chain %s: %s; retrying", key[1], key[0], exc) + self._retry[key] = (failures, self.now() + self.config.retry_initial) + except Exception as exc: # noqa: BLE001 - one vault's failure must not block the others + delay = backoff_delay(failures + 1, self.config.retry_initial, self.config.retry_max) + logger.warning( + "live evaluation for %s on chain %s failed (%s); retry in %.0fs", key[1], key[0], exc, delay + ) + self._retry[key] = (failures + 1, self.now() + delay) + else: + self._retry.pop(key, None) + return decided + + def _record(self, events: list[PendingEvent], outcome: str, reason: str) -> None: + """Record the same outcome for every event, with no alert requests.""" + decisions = [Decision(event, outcome, reason) for event in events] + self.store.record(decisions, [], self.config.mode, STATUS_OBSERVED, self.now()) + + def _evaluate_vault(self, events: list[PendingEvent]) -> int: + """Evaluate one vault's fresh events and store the decisions with their alert requests.""" + first = GovernanceEvent.from_payload(events[0].payload) + chain = Chain.from_chain_id(first.chain_id) + vault = vault_config(chain, first.vault, first.version) + if vault is None: + self._record(events, OUTCOME_NO_ALERT, "vault not in config.py") + return len(events) + governance_events = [GovernanceEvent.from_payload(e.payload) for e in events] + if self.config.mode == MODE_PRODUCTION: + # Same contract as the daily jobs: check, send, then record as alerted, all + # under the lock they take too, so one change is alerted once. + with governance_lock(vault.address): + evaluation = self.evaluate(chain, vault, first.version, governance_events) + rendered = render_vault_alerts(evaluation.header, evaluation.alerts) + for severity, text in rendered: + self.publish(severity, text) # raises on failure: nothing is committed or recorded + evaluation.commit() + else: + evaluation = self.evaluate(chain, vault, first.version, governance_events) + rendered = render_vault_alerts(evaluation.header, evaluation.alerts) + messages = [(severity.value, text) for severity, text in rendered] + decisions = [] + for event in events: + reason = evaluation.reasons.get(event.event_id, "not evaluated") + outcome = OUTCOME_ALERTED if reason == REASON_ALERTED and messages else OUTCOME_NO_ALERT + decisions.append(Decision(event, outcome, reason)) + status = {MODE_SHADOW: STATUS_PENDING, MODE_PRODUCTION: STATUS_SENT}.get(self.config.mode, STATUS_OBSERVED) + alert_ids = self.store.record(decisions, messages, self.config.mode, status, self.now()) + for severity, text in messages: + logger.info("live %s %s alert for %s: %s", self.config.mode, severity, vault.name, text.split("\n", 1)[0]) + if alert_ids: + logger.info("recorded %s live alert request(s) %s for %s", len(alert_ids), alert_ids, vault.name) + return len(events) + + async def _delivery_loop(self) -> None: + """Send pending shadow requests in order; on failure, retry the same request with backoff.""" + while True: + delay = self.config.delivery_interval + for request in await asyncio.to_thread(self.store.pending_alerts): + try: + await asyncio.to_thread(self.send, request) + except Exception as exc: # noqa: BLE001 - delivery is retried, never dropped + self._delivery_failures += 1 + await asyncio.to_thread(self.store.mark_failed, request.id, f"{type(exc).__name__}: {exc}") + delay = backoff_delay(self._delivery_failures, self.config.retry_initial, self.config.retry_max) + logger.warning("shadow delivery of request %s failed (%s); retry in %.0fs", request.id, exc, delay) + break # keep order: nothing after a failed request goes out first + self._delivery_failures = 0 + await asyncio.to_thread(self.store.mark_sent, request.id, self.now()) + await asyncio.sleep(delay) + + +def build_consumer(store: LiveStore, mode: str) -> LiveGovernanceConsumer: + """Return a consumer for ``mode``, downgrading shadow to observe when no errors chat is configured.""" + if mode == MODE_SHADOW and not error_channel_configured(): + logger.warning("MORPHO_LIVE_DELIVERY=shadow but no errors chat is configured; observing only") + mode = MODE_OBSERVE + return LiveGovernanceConsumer(store, LiveConfig(mode=mode)) + + +# Undecided events or undelivered shadow alerts older than this make `health` fail. +STALE_SECONDS = 600 + + +def live_health(store: LiveStore, now: float) -> list[str]: + """Return problems with the consumer's backlog: events not decided, or shadow alerts not delivered, in time.""" + problems = [] + stale_events = [e for e in store.undecided() if now - e.first_seen_at > STALE_SECONDS] + if stale_events: + problems.append(f"live: {len(stale_events)} event(s) undecided for over {STALE_SECONDS}s") + pending = store.pending_alerts() + if pending: + oldest = min(row["created_at"] for row in store.alerts_since(0) if row["status"] == STATUS_PENDING) + if now - oldest > STALE_SECONDS: + problems.append(f"live: {len(pending)} shadow alert(s) undelivered, oldest {now - oldest:.0f}s") + return problems + + +def live_summary(store: LiveStore, hours: float, now: float) -> str: + """Summarize the consumer's decisions and alert requests over the last ``hours``.""" + since = now - hours * 3600 + decisions = store.decisions_since(since) + alerts = store.alerts_since(since) + by_outcome: dict[str, int] = {} + for row in decisions: + by_outcome[row["outcome"]] = by_outcome.get(row["outcome"], 0) + 1 + delays = sorted( + row["decided_at"] - row["block_timestamp"] for row in decisions if row["outcome"] != OUTCOME_HISTORICAL + ) + by_status: dict[str, int] = {} + for row in alerts: + by_status[row["status"]] = by_status.get(row["status"], 0) + 1 + p95 = delays[max(0, -(-len(delays) * 95 // 100) - 1)] if delays else None + lines = [ + f"Live governance decisions, last {hours:g}h", + f" decisions: {len(decisions)} {dict(sorted(by_outcome.items()))}", + " block → decision: " + ("n/a" if p95 is None else f"p50 {delays[len(delays) // 2]:.1f}s, p95 {p95:.1f}s"), + f" alert requests: {len(alerts)} {dict(sorted(by_status.items()))}", + ] + for row in alerts[-10:]: + lines.append(f" - #{row['id']} {row['status']} {row['severity']}: {row['message'].split(chr(10), 1)[0][:120]}") + return "\n".join(lines) diff --git a/protocols/morpho/live_rules.py b/protocols/morpho/live_rules.py new file mode 100644 index 00000000..f10425c1 --- /dev/null +++ b/protocols/morpho/live_rules.py @@ -0,0 +1,288 @@ +"""Decide what the daily Morpho governance jobs would alert for newly indexed events. + +The indexer has no revoke/accept events for V1 and no ``Revoke`` for V2, so an +event alone cannot say whether its operation is still pending. Each event is +therefore a trigger: the vault's *current* on-chain state is read (at a block at +or after the event) and fed to the same check functions the daily +``governance.py`` / ``governance_v2.py`` jobs run. That gives the live path the +daily jobs' content, severity, grouping, loss classification, and dedup cache +(an operation the daily job already alerted is not alerted again), and an +operation revoked or executed before the read is simply not pending anymore. + +Cache writes buffered by those checks are returned as ``VaultEvaluation.commit`` +and never applied here. In observation and shadow mode nobody calls it, so the +daily jobs stay the only writers and keep alerting exactly as before; production +mode calls it once the alert is delivered, like the daily jobs do. Resolutions +(``Accept``; executed/revoked LOW notes) are left to the daily jobs. + +V1: ``SubmitCap`` / ``SubmitMarketRemoval`` → that market's pending cap and +removal checks; ``SubmitTimelock`` → the timelock (and guardian) check. +V2: ``Submit`` → ``executableAt(data)`` must still be set, then the pending +operation diff (loss alerts included); ``SetOwner`` / ``SetCurator`` → role diff; +``AddAdapter`` / ``RemoveAdapter`` → adapter set diff. +""" + +import json +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Any + +from web3 import Web3 + +from protocols.morpho import governance, governance_v2 +from protocols.morpho._alerts import VaultAlert, VaultDiff +from protocols.morpho.config import VAULTS_V1_BY_CHAIN, VAULTS_V2_BY_CHAIN, VaultConfig +from protocols.morpho.governance_v2 import PendingConfig, V2GovernanceSnapshot +from protocols.morpho.v2_decoders import selector_function_name +from utils.abi import load_abi +from utils.chains import Chain +from utils.erc20_metadata import fetch_erc20_metadata +from utils.logger import get_logger + +logger = get_logger("morpho.live_rules") + +ABI_VAULT_V2 = load_abi("protocols/morpho/abi/vault_v2.json") +# Any V2 adapter exposes realAssets(); this ABI carries it. +ABI_ADAPTER = load_abi("protocols/morpho/abi/morpho_vault_v1_adapter.json") + +V1_MARKET_EVENTS = frozenset({"SubmitCap", "SubmitMarketRemoval"}) +RESOLUTION_EVENTS = frozenset({"Accept"}) + +# Per-event outcome reasons recorded in the decision log. +REASON_ALERTED = "alerted" +REASON_NOT_NEW = "nothing new: no longer pending, or already alerted by the daily job" +REASON_RESOLUTION = "resolution: left to the daily job" +REASON_UNSUPPORTED = "event not handled by the live rules" + + +class RpcBehindError(Exception): + """The RPC node has not reached the event's block yet; retry shortly.""" + + +@dataclass(frozen=True) +class GovernanceEvent: + """One indexed ``MorphoGovernanceEvent`` occurrence.""" + + event_id: str + chain_id: int + vault: str # checksummed + version: int + name: str + block_number: int + transaction_hash: str + args: dict[str, Any] + + @classmethod + def from_payload(cls, payload: dict[str, Any]) -> "GovernanceEvent": + """Build from an inbox row's stored Envio payload.""" + try: + args = json.loads(payload.get("argsJson") or "{}") + except TypeError, ValueError: + args = {} + return cls( + event_id=str(payload["id"]), + chain_id=int(payload["chainId"]), + vault=str(Web3.to_checksum_address(payload["vaultAddress"])), + version=int(payload.get("vaultVersion") or 0), + name=str(payload.get("eventName") or ""), + block_number=int(payload["blockNumber"]), + transaction_hash=str(payload.get("transactionHash") or ""), + args=args if isinstance(args, dict) else {}, + ) + + +@dataclass +class VaultEvaluation: + """What the daily jobs would alert for one vault's batch of events.""" + + header: str + alerts: list[VaultAlert] = field(default_factory=list) + # event_id -> REASON_* explaining the event's outcome + reasons: dict[str, str] = field(default_factory=dict) + # Applies the daily jobs' cache writes for these alerts; call only after delivery. + commit: Callable[[], None] = field(default=lambda: None, compare=False) + + +def vault_config(chain: Chain, address: str, version: int) -> VaultConfig | None: + """Return the configured vault for ``address`` on ``chain`` and generation ``version``.""" + vaults = (VAULTS_V1_BY_CHAIN if version == 1 else VAULTS_V2_BY_CHAIN).get(chain, ()) + return next((v for v in vaults if v.address.lower() == address.lower()), None) + + +def _current_block(client: Any, events: list[GovernanceEvent]) -> int: + """Return the RPC head, or raise ``RpcBehindError`` if it trails the newest event.""" + block = int(client.eth.block_number) + newest = max(e.block_number for e in events) + if block < newest: + raise RpcBehindError(f"RPC head {block} is behind event block {newest}") + return block + + +def _hex_bytes(value: Any) -> bytes: + """Decode a ``bytes`` or ``bytes32`` argument from ``argsJson`` (0x hex).""" + return bytes.fromhex(str(value).removeprefix("0x")) + + +def _attribute(diff: VaultDiff, before: int, ids: list[str], reasons: dict[str, str]) -> None: + """Record ``ids`` as alerted when the check added sections since ``before``, else as not new.""" + reason = REASON_ALERTED if len(diff.alerts) > before else REASON_NOT_NEW + for event_id in ids: + reasons[event_id] = reason + + +def evaluate_v1(client: Any, chain: Chain, vault: VaultConfig, events: list[GovernanceEvent]) -> VaultEvaluation: + """Run the daily V1 checks the events point at, against the vault's current state.""" + block = _current_block(client, events) + contract = client.eth.contract(address=Web3.to_checksum_address(vault.address), abi=governance.ABI_MORPHO) + diff = VaultDiff() + result = VaultEvaluation(header=governance.vault_header(vault.name, vault.address, chain)) + + by_market: dict[bytes, list[str]] = {} + timelock_ids: list[str] = [] + for event in events: + if event.name in V1_MARKET_EVENTS and event.args.get("id"): + by_market.setdefault(_hex_bytes(event.args["id"]), []).append(event.event_id) + elif event.name == "SubmitTimelock": + timelock_ids.append(event.event_id) + else: + result.reasons[event.event_id] = REASON_UNSUPPORTED + + if by_market: + states = governance.load_market_governance_states(contract, list(by_market), client, block) + for market_id, state in zip(by_market, states, strict=True): + before = len(diff.alerts) + # One check per market: a SubmitCap and a SubmitMarketRemoval on the same + # market in one batch must not render its sections twice. + governance.check_market_governance_state(vault.name, state, chain, diff) + _attribute(diff, before, by_market[market_id], result.reasons) + if timelock_ids: + before = len(diff.alerts) + governance.check_timelock_and_guardian(contract, client, diff, block) + _attribute(diff, before, timelock_ids, result.reasons) + + result.alerts = diff.alerts + result.commit = diff.commit + return result + + +def load_v2_snapshot(client: Any, chain: Chain, vault: VaultConfig, block: int) -> V2GovernanceSnapshot: + """Read the parts of a V2 vault's governance snapshot the live rules need, on-chain. + + The Morpho API lags the chain, so the daily job's GraphQL snapshot cannot be + used for a seconds-old event. Sentinels and allocators are mappings (not + enumerable on-chain) and are not needed: no indexed event changes them. + """ + address = Web3.to_checksum_address(vault.address) + contract = client.eth.contract(address=address, abi=ABI_VAULT_V2) + with client.batch_requests() as batch: + batch.add(contract.functions.owner().call(block_identifier=block)) + batch.add(contract.functions.curator().call(block_identifier=block)) + batch.add(contract.functions.asset().call(block_identifier=block)) + batch.add(contract.functions.adaptersLength().call(block_identifier=block)) + owner, curator, asset, length = client.execute_batch(batch) + adapters: list[str] = [] + if length: + with client.batch_requests() as batch: + for index in range(int(length)): + batch.add(contract.functions.adapters(index).call(block_identifier=block)) + adapters = [str(Web3.to_checksum_address(a)) for a in client.execute_batch(batch)] + + # Adapter valuations decide whether a removal is a potential loss, so a failed + # read must not look like an empty adapter: it raises, and the vault is retried. + adapter_assets: dict[str, int] = {} + if adapters: + with client.batch_requests() as batch: + for adapter in adapters: + batch.add(_adapter(client, adapter).functions.realAssets().call(block_identifier=block)) + adapter_assets = {a: int(v) for a, v in zip(adapters, client.execute_batch(batch), strict=True)} + + metadata = fetch_erc20_metadata(chain.chain_id, str(asset)) + return V2GovernanceSnapshot( + name=vault.name, + address=str(address), + chain=chain, + owner=str(Web3.to_checksum_address(owner)), + curator=str(Web3.to_checksum_address(curator)), + sentinels=[], + allocators=[], + adapters=sorted(adapters), + asset_symbol=metadata.symbol if metadata else "", + asset_decimals=metadata.decimals if metadata else None, + total_assets=_try_call(contract.functions.totalAssets(), block), + adapter_assets=adapter_assets, + ) + + +def _adapter(client: Any, address: str) -> Any: + """Return an adapter contract exposing ``realAssets()``.""" + return client.eth.contract(address=address, abi=ABI_ADAPTER) + + +def _try_call(function: Any, block: int) -> int | None: + """Best-effort integer view call for display-only values; None when it reverts (e.g. an IRM that reverts).""" + try: + return int(function.call(block_identifier=block)) + except Exception as e: # noqa: BLE001 - a reverting valuation only drops the amount from the alert + logger.warning("view call failed, alert omits the amount: %s", e) + return None + + +def evaluate_v2(client: Any, chain: Chain, vault: VaultConfig, events: list[GovernanceEvent]) -> VaultEvaluation: + """Run the daily V2 diff for the events' operations, roles, and adapters, against current state.""" + block = _current_block(client, events) + snapshot = load_v2_snapshot(client, chain, vault, block) + diff = VaultDiff() + result = VaultEvaluation(header=governance_v2.vault_header(snapshot)) + + submits = [e for e in events if e.name == "Submit" and e.args.get("data")] + if submits: + contract = client.eth.contract(address=snapshot.address, abi=ABI_VAULT_V2) + datas = [_hex_bytes(e.args["data"]) for e in submits] + with client.batch_requests() as batch: + for data in datas: + batch.add(contract.functions.executableAt(data).call(block_identifier=block)) + executable_at = client.execute_batch(batch) + new_pending: list[tuple[PendingConfig, str]] = [] + for event, data, valid_at in zip(submits, datas, executable_at, strict=True): + if not valid_at: + result.reasons[event.event_id] = REASON_NOT_NEW # accepted or revoked already + continue + pc = PendingConfig( + valid_at=int(valid_at), + function_name=selector_function_name(data[:4]) or "", + data=data, + tx_hash=event.transaction_hash, + ) + before = len(diff.alerts) + new = governance_v2.diff_pending_config(snapshot, pc, diff) + if new is not None: + new_pending.append(new) + result.reasons[event.event_id] = ( + REASON_ALERTED if new is not None or len(diff.alerts) > before else REASON_NOT_NEW + ) + # Same grouping as the daily job: one section for every new non-loss submission. + governance_v2.alert_pending_new(snapshot, new_pending, diff) + + checks: list[tuple[set[str], Callable[[], None]]] = [ + ({"SetOwner"}, lambda: governance_v2.diff_single_role(snapshot, "owner", snapshot.owner, diff)), + ({"SetCurator"}, lambda: governance_v2.diff_single_role(snapshot, "curator", snapshot.curator, diff)), + ( + {"AddAdapter", "RemoveAdapter"}, + lambda: governance_v2.diff_set(snapshot, "adapters", snapshot.adapters, diff), + ), + ] + for names, check in checks: + ids = [e.event_id for e in events if e.name in names] + if ids: + before = len(diff.alerts) + check() + _attribute(diff, before, ids, result.reasons) + + for event in events: + if event.event_id not in result.reasons: + result.reasons[event.event_id] = ( + REASON_RESOLUTION if event.name in RESOLUTION_EVENTS else REASON_UNSUPPORTED + ) + result.alerts = diff.alerts + result.commit = diff.commit + return result diff --git a/protocols/morpho/live_store.py b/protocols/morpho/live_store.py new file mode 100644 index 00000000..056aafb5 --- /dev/null +++ b/protocols/morpho/live_store.py @@ -0,0 +1,309 @@ +"""Durable decisions and alert requests for the live Morpho governance consumer. + +Two tables live in the Morpho stream's inbox file (``morpho_subscriber.db``), +next to the events they are about but written by the consumer, never by +ingestion: + +- ``morpho_live_decisions``: one row per evaluated event revision with its + outcome (``alerted``, ``no_alert``, ``historical``) and the reason. An event is + pending evaluation until it has a decision for its current revision, so a reorg + that rewrites an event re-evaluates it, and a restart resumes where it stopped. +- ``morpho_live_alerts``: one row per message the daily jobs would have sent, + with its delivery mode and status. Observation mode records ``observed``; + shadow mode records ``pending`` and the delivery loop moves it to ``sent`` (or + keeps retrying, recording ``last_error``); production records ``sent``. Each + request keeps the revisions of the events it came from: a pending request whose + events were removed or rewritten by a reorg is ``invalidated`` instead of sent + (the rewritten event is evaluated again and makes its own request). +""" + +import json +import sqlite3 +from collections.abc import Iterator +from contextlib import closing, contextmanager +from dataclasses import dataclass +from typing import Any + +from utils.subscriber.inbox import EventInbox + +OUTCOME_ALERTED = "alerted" +OUTCOME_NO_ALERT = "no_alert" +OUTCOME_HISTORICAL = "historical" + +STATUS_OBSERVED = "observed" +STATUS_PENDING = "pending" +STATUS_SENT = "sent" +STATUS_INVALIDATED = "invalidated" + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS morpho_live_decisions ( + event_id TEXT PRIMARY KEY, + revision INTEGER NOT NULL, + chain_id INTEGER NOT NULL, + vault_address TEXT NOT NULL, + event_name TEXT, + block_timestamp INTEGER NOT NULL, + first_seen_at REAL NOT NULL, + decided_at REAL NOT NULL, + outcome TEXT NOT NULL, + reason TEXT, + alert_ids TEXT +); +CREATE INDEX IF NOT EXISTS idx_morpho_live_decisions_decided ON morpho_live_decisions(decided_at); + +CREATE TABLE IF NOT EXISTS morpho_live_alerts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + chain_id INTEGER NOT NULL, + vault_address TEXT NOT NULL, + mode TEXT NOT NULL, + severity TEXT NOT NULL, + message TEXT NOT NULL, + event_ids TEXT NOT NULL, + -- {event_id: revision} of the events the request came from + event_revisions TEXT NOT NULL, + created_at REAL NOT NULL, + status TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + sent_at REAL +); +CREATE INDEX IF NOT EXISTS idx_morpho_live_alerts_status ON morpho_live_alerts(status, id); +""" + + +@dataclass(frozen=True) +class PendingEvent: + """An inbox event whose current revision has no decision yet.""" + + event_id: str + revision: int + chain_id: int + vault_address: str + event_name: str | None + block_timestamp: int + first_seen_at: float + payload: dict[str, Any] + + +@dataclass(frozen=True) +class Decision: + """The outcome for one event revision.""" + + event: PendingEvent + outcome: str + reason: str + + +@dataclass(frozen=True) +class AlertRequest: + """One rendered message the daily jobs would send for a vault.""" + + id: int + chain_id: int + vault_address: str + mode: str + severity: str + message: str + event_ids: list[str] + attempts: int + + +class LiveStore: + """SQLite store for the consumer's decisions and alert requests, in the stream inbox's file.""" + + def __init__(self, inbox: EventInbox) -> None: + self.inbox = inbox + self.path = inbox.path + self._events = inbox.stream.table("events") + with self._connect(): + pass + + @contextmanager + def _connect(self) -> Iterator[sqlite3.Connection]: + """Open a short-lived connection; one transaction spans the ``with`` block.""" + with closing(sqlite3.connect(self.path, timeout=10)) as conn: + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=10000") + conn.executescript(_SCHEMA) + with conn: + yield conn + + def undecided(self, limit: int = 500, exclude: set[tuple[int, str]] | None = None) -> list[PendingEvent]: + """Return active events without a decision for their current revision, in chain order. + + Args: + limit: Most rows to return. + exclude: ``(chain id, lowercase vault)`` pairs to skip (vaults in backoff), + applied before ``limit`` so they cannot crowd out other vaults. + """ + skip = sorted(exclude or ()) + not_excluded = "".join(" AND NOT (e.chain_id = ? AND e.contract_address = ?)" for _ in skip) + with self._connect() as conn: + rows = conn.execute( + f""" + SELECT e.* FROM {self._events} e + LEFT JOIN morpho_live_decisions d ON d.event_id = e.event_id AND d.revision = e.revision + WHERE e.status = 'active' AND d.event_id IS NULL{not_excluded} + ORDER BY e.chain_id, e.block_number, e.log_index + LIMIT ? + """, + (*[value for pair in skip for value in pair], limit), + ).fetchall() + return [ + PendingEvent( + event_id=row["event_id"], + revision=row["revision"], + chain_id=row["chain_id"], + vault_address=row["contract_address"], + event_name=row["event_name"], + block_timestamp=row["block_timestamp"], + first_seen_at=row["first_seen_at"], + payload=json.loads(row["payload_json"]), + ) + for row in rows + ] + + def record( + self, + decisions: list[Decision], + messages: list[tuple[str, str]], + mode: str, + status: str, + now: float, + ) -> list[int]: + """Store one vault batch's decisions and its alert requests in one transaction. + + Args: + decisions: Outcome per event of the batch. + messages: ``(severity, message)`` pairs to request, in send order. + mode: Delivery mode the requests were made under. + status: ``STATUS_OBSERVED`` or ``STATUS_PENDING``. + now: Decision time (unix seconds). + + Returns: + The new alert request ids. + """ + alerted = [d.event for d in decisions if d.outcome == OUTCOME_ALERTED] + alerted_ids = json.dumps([e.event_id for e in alerted]) + revisions = json.dumps({e.event_id: e.revision for e in alerted}) + with self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") + alert_ids = [] + if messages: + chain_id, vault = decisions[0].event.chain_id, decisions[0].event.vault_address + for severity, message in messages: + cursor = conn.execute( + """ + INSERT INTO morpho_live_alerts ( + chain_id, vault_address, mode, severity, message, event_ids, event_revisions, + created_at, status + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + (chain_id, vault, mode, severity, message, alerted_ids, revisions, now, status), + ) + alert_ids.append(int(cursor.lastrowid or 0)) + for decision in decisions: + event = decision.event + conn.execute( + """ + INSERT INTO morpho_live_decisions ( + event_id, revision, chain_id, vault_address, event_name, block_timestamp, first_seen_at, + decided_at, outcome, reason, alert_ids + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(event_id) DO UPDATE SET + revision = excluded.revision, decided_at = excluded.decided_at, outcome = excluded.outcome, + reason = excluded.reason, alert_ids = excluded.alert_ids + """, + ( + event.event_id, + event.revision, + event.chain_id, + event.vault_address, + event.event_name, + event.block_timestamp, + event.first_seen_at, + now, + decision.outcome, + decision.reason, + json.dumps(alert_ids if decision.outcome == OUTCOME_ALERTED else []), + ), + ) + return alert_ids + + def pending_alerts(self, limit: int = 50) -> list[AlertRequest]: + """Return alert requests awaiting delivery, oldest first, invalidating reorged ones. + + A request whose source events are no longer active at the revisions it was + made from (removed, or rewritten and so evaluated again) is marked + ``invalidated`` and not returned. + """ + with self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") + rows = conn.execute( + "SELECT * FROM morpho_live_alerts WHERE status = ? ORDER BY id LIMIT ?", (STATUS_PENDING, limit) + ).fetchall() + current = [] + for row in rows: + if self._sources_current(conn, json.loads(row["event_revisions"])): + current.append(row) + else: + conn.execute( + "UPDATE morpho_live_alerts SET status = ?, last_error = ? WHERE id = ?", + (STATUS_INVALIDATED, "source event removed or rewritten by a reorg", row["id"]), + ) + rows = current + return [ + AlertRequest( + id=row["id"], + chain_id=row["chain_id"], + vault_address=row["vault_address"], + mode=row["mode"], + severity=row["severity"], + message=row["message"], + event_ids=json.loads(row["event_ids"]), + attempts=row["attempts"], + ) + for row in rows + ] + + def _sources_current(self, conn: sqlite3.Connection, revisions: dict[str, int]) -> bool: + """True when every source event is still active at the revision the request was made from.""" + for event_id, revision in revisions.items(): + row = conn.execute( + f"SELECT status, revision FROM {self._events} WHERE event_id = ?", (event_id,) + ).fetchone() + if row is None or row["status"] != "active" or row["revision"] != revision: + return False + return True + + def mark_sent(self, alert_id: int, now: float) -> None: + """Record a delivered request.""" + with self._connect() as conn: + conn.execute( + "UPDATE morpho_live_alerts SET status = ?, attempts = attempts + 1, sent_at = ?, last_error = NULL " + "WHERE id = ?", + (STATUS_SENT, now, alert_id), + ) + + def mark_failed(self, alert_id: int, error: str) -> None: + """Record a failed delivery attempt; the request stays pending and is retried.""" + with self._connect() as conn: + conn.execute( + "UPDATE morpho_live_alerts SET attempts = attempts + 1, last_error = ? WHERE id = ?", + (error[:500], alert_id), + ) + + def decisions_since(self, since: float) -> list[sqlite3.Row]: + """Return decisions made at or after ``since``, oldest first.""" + with self._connect() as conn: + return conn.execute( + "SELECT * FROM morpho_live_decisions WHERE decided_at >= ? ORDER BY decided_at, event_id", (since,) + ).fetchall() + + def alerts_since(self, since: float) -> list[sqlite3.Row]: + """Return alert requests created at or after ``since``, oldest first.""" + with self._connect() as conn: + return conn.execute( + "SELECT * FROM morpho_live_alerts WHERE created_at >= ? ORDER BY id", (since,) + ).fetchall() diff --git a/protocols/timelock/event_subscriber.py b/protocols/timelock/event_subscriber.py index c5524bb2..d39cd3c5 100644 --- a/protocols/timelock/event_subscriber.py +++ b/protocols/timelock/event_subscriber.py @@ -31,7 +31,11 @@ from dotenv import load_dotenv +# Before anything imports utils.cache: live Morpho checks must use the daily jobs' cache namespace. +from protocols.morpho.daily_governance_cache import check_daily_governance_cache # isort: skip from protocols.morpho.governance_stream import MORPHO_GOVERNANCE_STREAM, morpho_governance_scope +from protocols.morpho.live_governance import DELIVERY_MODES, MODE_SHADOW, build_consumer, live_health, live_summary +from protocols.morpho.live_store import LiveStore from protocols.timelock.event_stream import TIMELOCK_STREAM, timelock_scope from utils.graphql_ws import http_to_ws_url from utils.logger import get_logger @@ -118,14 +122,21 @@ def cmd_run(args: argparse.Namespace) -> None: inbox = EventInbox(stream) notifier = build_notifier(args.notify, stream.name) notices = PilotNotices(notifier, inbox, scope.labels) if notifier is not None else None - workers.append(Worker(post, ws_url, inbox, scope.addresses, config, notices=notices, heads=heads)) + consumer = None + if stream is MORPHO_GOVERNANCE_STREAM: + check_daily_governance_cache() + consumer = build_consumer(LiveStore(inbox), args.morpho_delivery) + workers.append( + Worker(post, ws_url, inbox, scope.addresses, config, notices=notices, heads=heads, consumer=consumer) + ) _logger.info( - "starting %s stream: %s chains, transport=%s reconcile every %.0fs notices=%s", + "starting %s stream: %s chains, transport=%s reconcile every %.0fs notices=%s live=%s", stream.name, len(scope.addresses), config.transport, config.reconcile_interval, "errors chat" if notices else "off", + consumer.config.mode if consumer else "off", ) asyncio.run(run_workers(workers, watchdog_interval=config.heartbeat_interval)) @@ -155,8 +166,11 @@ def cmd_report(args: argparse.Namespace) -> None: reports = {stream.name: build_report(EventInbox(stream), args.hours) for stream, _ in _selected(args)} if args.json: print(json.dumps({name: report_to_json(r) for name, r in reports.items()}, indent=2)) - else: - print("\n\n".join(format_report(r) for r in reports.values())) + return + sections = [format_report(r) for r in reports.values()] + if MORPHO_GOVERNANCE_STREAM.name in reports: + sections.append(live_summary(LiveStore(EventInbox(MORPHO_GOVERNANCE_STREAM)), args.hours, time.time())) + print("\n\n".join(sections)) def heartbeat_problems(status: Mapping[str, tuple[Any, float]], now: float) -> list[str]: @@ -182,8 +196,11 @@ def cmd_health(args: argparse.Namespace) -> None: streams = {} problems: list[str] = [] for stream, _ in _selected(args): - status = EventInbox(stream).status() + inbox = EventInbox(stream) + status = inbox.status() stream_problems = heartbeat_problems(status, now) + if stream is MORPHO_GOVERNANCE_STREAM: + stream_problems += live_health(LiveStore(inbox), now) problems += [f"{stream.name}: {problem}" for problem in stream_problems] streams[stream.name] = {"problems": stream_problems, "heartbeat": status.get("heartbeat", (None, 0.0))[0]} print(json.dumps({"ok": not problems, "problems": problems, "streams": streams}, indent=2)) @@ -226,6 +243,15 @@ def add_reconcile_args(p: argparse.ArgumentParser) -> None: default=os.getenv("TIMELOCK_SUBSCRIBER_NOTIFY", "none"), help="Post pilot notices to the internal errors chat (errors) or not at all (none, default).", ) + run.add_argument( + "--morpho-delivery", + choices=DELIVERY_MODES, + default=os.getenv("MORPHO_LIVE_DELIVERY", MODE_SHADOW), + help=( + "Live Morpho governance alerts: post to the errors chat (shadow, default; observe without one), " + "record only (observe), or send to the Morpho channel (production)." + ), + ) add_reconcile_args(run) run.set_defaults(func=cmd_run) diff --git a/tests/test_event_subscriber_streams.py b/tests/test_event_subscriber_streams.py index 3143d3ef..c80a855e 100644 --- a/tests/test_event_subscriber_streams.py +++ b/tests/test_event_subscriber_streams.py @@ -11,14 +11,12 @@ import pytest -from protocols.morpho.config import VAULTS_V1_BY_CHAIN, VAULTS_V2_BY_CHAIN, iter_vaults from protocols.morpho.governance_stream import ( MORPHO_GOVERNANCE_STREAM, describe_governance_event, - morpho_governance_scope, ) from protocols.timelock.event_stream import TIMELOCK_STREAM -from protocols.timelock.event_subscriber import STREAMS, heartbeat_problems, parse_streams +from protocols.timelock.event_subscriber import heartbeat_problems, parse_streams from tests.timelock_subscriber_fakes import ( MORPHO_VAULT, TIMELOCK, @@ -28,12 +26,12 @@ make_morpho_event, ) from utils.subscriber import envio as envio_module -from utils.subscriber.envio import EnvioQueryError, SharedChainHeads, http_poster, live_subscription, range_query -from utils.subscriber.inbox import SOURCE_HTTP, SOURCE_WS, EventInbox +from utils.subscriber.envio import EnvioQueryError, SharedChainHeads, http_poster +from utils.subscriber.inbox import SOURCE_WS, EventInbox from utils.subscriber.notices import NoticeConfig, PilotNotices from utils.subscriber.notifier import Notifier from utils.subscriber.reconcile import ReconcileConfig, reconcile_all -from utils.subscriber.report import build_report, format_report +from utils.subscriber.stream import StreamLabel from utils.subscriber.worker import Worker, WorkerConfig, run_workers TIMELOCK_SCOPE = {1: {TIMELOCK}} @@ -65,32 +63,6 @@ def _ids(inbox: EventInbox) -> set[str]: # ── stream definitions ───────────────────────────────────────────────── -def test_morpho_queries_target_the_governance_entity() -> None: - query = range_query(MORPHO_GOVERNANCE_STREAM, 1, 10, 20, 50, 0, since_ts=None) - assert "MorphoGovernanceEvent(" in query and "vaultAddress: { _in: $addresses }" in query - assert "argsJson" in query and "timelockAddress" not in query - live = live_subscription(MORPHO_GOVERNANCE_STREAM, 50, [8453, 1]) - assert "subscription MorphoGovernanceEventLive" in live and "chainId: { _in: [1, 8453] }" in live - - -def test_morpho_scope_covers_every_configured_vault_with_a_label() -> None: - scope = morpho_governance_scope() - configured = [*iter_vaults(VAULTS_V1_BY_CHAIN), *iter_vaults(VAULTS_V2_BY_CHAIN)] - - assert sum(len(addresses) for addresses in scope.addresses.values()) == len(configured) - for chain, vault in configured: - assert vault.address.lower() in scope.addresses[chain.chain_id] - assert scope.labels[(vault.address.lower(), chain.chain_id)].protocol == "morpho" - assert scope.labels[(MORPHO_VAULT, 1)].label == "Yearn USDC V1" - - -def test_stream_names_tables_and_files_are_distinct() -> None: - assert TIMELOCK_STREAM.table("events") == "timelock_sub_events" # unchanged for the running pilot - assert MORPHO_GOVERNANCE_STREAM.table("events") == "morpho_sub_events" - assert TIMELOCK_STREAM.db_filename != MORPHO_GOVERNANCE_STREAM.db_filename - assert set(STREAMS) == {"timelock", "morpho"} - - def test_describe_governance_event_shows_args_and_survives_bad_json() -> None: lines = describe_governance_event(make_morpho_event(1, 10, vaultVersion=2, argsJson='{"b":"2","a":true}')) assert lines == ["Vault V2, sent by 0x" + "cd" * 20, 'Args: {"a":true,"b":"2"}'] @@ -186,30 +158,25 @@ def fake_request(*_args: object, session: object, **_kwargs: object) -> Response assert sessions[2] is not sessions[0] -# ── notices and reports ──────────────────────────────────────────────── +# ── notices ──────────────────────────────────────────────── def test_morpho_notice_names_the_vault_and_its_arguments(tmp_path: Path) -> None: _, inbox = _inboxes(tmp_path) notifier = RecordingNotifier() - notices = PilotNotices(notifier, inbox, morpho_governance_scope().labels, NoticeConfig(), now=lambda: 0.0) + labels = {(MORPHO_VAULT, 1): StreamLabel("Test Vault V1", "morpho")} + notices = PilotNotices(notifier, inbox, labels, NoticeConfig(), now=lambda: 0.0) inbox.observe([make_morpho_event(1, 26_000_000, 7)], SOURCE_WS, 1_700_000_005.0) asyncio.run(notices.new_events(SOURCE_WS, ["1_26000000_7"])) (text,) = notifier.posted assert text.startswith("🛰️ New Morpho governance event via WS, 5s after its block") - assert "Yearn USDC V1 (morpho) on Mainnet: SubmitMarketRemoval 1_26000000_7" in text + assert "Test Vault V1 (morpho) on Mainnet: SubmitMarketRemoval 1_26000000_7" in text assert f"Vault: {MORPHO_VAULT} https://etherscan.io/address/{MORPHO_VAULT}" in text assert "Args: {" in text and "Operation ID" not in text -def test_report_is_titled_by_stream(tmp_path: Path) -> None: - timelock_inbox, morpho_inbox = _inboxes(tmp_path) - assert format_report(build_report(timelock_inbox, 1)).startswith("Timelock subscriber pilot report") - assert format_report(build_report(morpho_inbox, 1)).startswith("Morpho governance subscriber pilot report") - - # ── one process, several streams ─────────────────────────────────────── @@ -449,20 +416,6 @@ async def crash() -> None: asyncio.run(scenario()) -def test_http_observation_of_both_streams_is_recorded_per_source(tmp_path: Path) -> None: - envio = FakeEnvio() - envio.set_head(1, 1) - timelock_inbox, morpho_inbox = _inboxes(tmp_path) - for inbox, scope in ((timelock_inbox, TIMELOCK_SCOPE), (morpho_inbox, MORPHO_SCOPE)): - reconcile_all(envio.post, inbox, scope, CONFIG) - envio.add(make_event(1, 10), make_morpho_event(1, 10, 1)) - for inbox, scope in ((timelock_inbox, TIMELOCK_SCOPE), (morpho_inbox, MORPHO_SCOPE)): - reconcile_all(envio.post, inbox, scope, CONFIG) - - assert {row["first_source"] for row in morpho_inbox.events_since(0)} == {SOURCE_HTTP} - assert {q["entity"] for q in envio.queries} == {"TimelockEvent", "MorphoGovernanceEvent"} - - # ── CLI ──────────────────────────────────────────────────────────────── diff --git a/tests/test_morpho_governance.py b/tests/test_morpho_governance.py index 8ded03bd..dbd3c380 100644 --- a/tests/test_morpho_governance.py +++ b/tests/test_morpho_governance.py @@ -232,7 +232,7 @@ def batch_requests() -> Any: client.batch_requests.side_effect = batch_requests client.execute_batch.side_effect = lambda batch: batch.calls - [state] = governance._load_market_governance_states(vault, [market_id], client, block_number=1) + [state] = governance.load_market_governance_states(vault, [market_id], client, block_number=1) self.assertIsNone(state.vault_total_assets) self.assertEqual( @@ -281,7 +281,7 @@ def test_findings_for_one_vault_collapse_into_a_single_message(self) -> None: for state in states: governance.check_market_governance_state("Yearn OG USDC", state, Chain.BASE, diff) governance.send_vault_alerts( - governance._vault_header("Yearn OG USDC", VAULT, Chain.BASE), + governance.vault_header("Yearn OG USDC", VAULT, Chain.BASE), diff.alerts, governance.PROTOCOL, ) diff --git a/tests/test_morpho_live_governance.py b/tests/test_morpho_live_governance.py new file mode 100644 index 00000000..4ce7d1b6 --- /dev/null +++ b/tests/test_morpho_live_governance.py @@ -0,0 +1,832 @@ +"""Tests for the live Morpho governance consumer (#448): rules parity, durable decisions, and delivery.""" + +import asyncio +import json +import time +from contextlib import ExitStack, contextmanager +from pathlib import Path +from typing import Any, Iterator +from unittest.mock import MagicMock, patch + +import pytest +from eth_abi import encode as abi_encode +from web3 import Web3 + +from protocols.morpho import governance_v2, live_rules +from protocols.morpho._alerts import LOSS_PREFIX, VaultAlert, render_vault_alerts +from protocols.morpho.config import VaultConfig +from protocols.morpho.daily_governance_cache import check_daily_governance_cache, daily_morpho_filename +from protocols.morpho.governance import MarketGovernanceState +from protocols.morpho.governance_stream import MORPHO_GOVERNANCE_STREAM +from protocols.morpho.governance_v2 import ROLE_TYPE, V2GovernanceSnapshot +from protocols.morpho.live_governance import ( + BASELINE_STATUS_KEY, + MODE_PRODUCTION, + MODE_SHADOW, + SHADOW_LABEL, + LiveConfig, + LiveGovernanceConsumer, + NotDelivered, + ShadowChannelMissing, + build_consumer, + live_health, + live_summary, + publish_to_morpho, + send_shadow, + shadow_texts, +) +from protocols.morpho.live_rules import ( + REASON_ALERTED, + REASON_NOT_NEW, + REASON_RESOLUTION, + GovernanceEvent, + RpcBehindError, + VaultEvaluation, + evaluate_v1, + evaluate_v2, +) +from protocols.morpho.live_store import ( + OUTCOME_ALERTED, + OUTCOME_HISTORICAL, + OUTCOME_NO_ALERT, + STATUS_INVALIDATED, + STATUS_OBSERVED, + STATUS_PENDING, + STATUS_SENT, + AlertRequest, + LiveStore, +) +from protocols.morpho.v2_decoders import selector_of +from tests.timelock_subscriber_fakes import FakeEnvio, FakeWSServer, make_morpho_event +from utils.alert import AlertSeverity +from utils.cache import morpho_key +from utils.chains import Chain +from utils.file_lock import LockTimeout, file_lock +from utils.subscriber.inbox import SOURCE_HTTP, SOURCE_WS, EventInbox +from utils.subscriber.reconcile import ReconcileConfig +from utils.subscriber.worker import Worker, WorkerConfig +from utils.telegram import MAX_MESSAGE_LENGTH + +V1_VAULT = "0x68Aea7b82Df6CcdF76235D46445Ed83f85F845A3" # Yearn USDC V1, mainnet +V2_VAULT = "0xaA8d9E2aBa210639cE6C7cE21385e7c673ACa6f3" # Yearn USDC V2, mainnet +ADAPTER = "0x" + "aa" * 20 +NEW_OWNER = "0x" + "bb" * 20 +MARKET = "0x" + "ab" * 32 +METADATA = {"name": "WETH/USDC", "lltv": 86 * 10**16, "loan_decimals": 6, "loan_symbol": "USDC"} +OTHER_VAULT = "0x" + "f9" * 20 +V1_CONFIG = VaultConfig("Yearn USDC", V1_VAULT, 1) +V2_CONFIG = VaultConfig("Yearn USDC", V2_VAULT, 1) +VALID_AT = 2_000_000_000 + + +@pytest.fixture(autouse=True) +def configured_vaults() -> Iterator[None]: + """Resolve vaults from synthetic configs, so the consumer tests don't depend on config.py's vault list.""" + configs = {V1_VAULT.lower(): V1_CONFIG, OTHER_VAULT: VaultConfig("Other", OTHER_VAULT, 1)} + with patch( + "protocols.morpho.live_governance.vault_config", + side_effect=lambda chain, address, version: configs.get(address.lower()), + ): + yield + + +def _calldata(sig: str, types: list[str], values: list[Any]) -> str: + return "0x" + (selector_of(sig) + abi_encode(types, values)).hex() + + +def _event(name: str, args: dict[str, Any], block: int = 100, log: int = 0, vault: str = V1_VAULT, version: int = 1): + payload = make_morpho_event( + 1, block, log, address=vault, vaultVersion=version, eventName=name, argsJson=json.dumps(args) + ) + return GovernanceEvent.from_payload(payload) + + +class _Batch(list): + def add(self, value: Any) -> None: + self.append(value) + + +def _client(head: int = 1_000) -> MagicMock: + """Client whose batch returns the values added to it (each fake ``call`` returns its value directly).""" + client = MagicMock() + client.eth.block_number = head + + @contextmanager + def batch_requests() -> Iterator[_Batch]: + yield _Batch() + + client.batch_requests.side_effect = batch_requests + client.execute_batch.side_effect = lambda batch: list(batch) + return client + + +@contextmanager +def _v1_cache(values: dict[tuple[str, str], int] | None = None) -> Iterator[MagicMock]: + """Patch the V1 cache (reads from ``values``, writes recorded) and market metadata.""" + values = values or {} + write = MagicMock() + with ExitStack() as stack: + stack.enter_context( + patch( + "protocols.morpho.governance.get_last_executed_morpho_from_file", + side_effect=lambda vault, market, kind: values.get((market, kind), 0), + ) + ) + stack.enter_context(patch("protocols.morpho.governance.write_last_executed_morpho_to_file", write)) + stack.enter_context(patch("protocols.morpho.governance.fetch_market_metadata", return_value=METADATA)) + stack.enter_context( + patch("protocols.morpho.governance.fetch_market_info", return_value=("WETH/USDC (86.00%)", 6)) + ) + yield write + + +def _state(**overrides: Any) -> MarketGovernanceState: + fields: dict[str, Any] = { + "vault_address": V1_VAULT, + "market_id": MARKET, + "pending_cap": 0, + "pending_cap_timestamp": 0, + "current_cap": 0, + "removable_at": 0, + } + fields.update(overrides) + return MarketGovernanceState(**fields) + + +# ── V1 rules ─────────────────────────────────────────────────────────── + + +def test_v1_submit_cap_alerts_like_the_daily_job_without_writing_its_cache() -> None: + state = _state(pending_cap=5 * 10**12, pending_cap_timestamp=VALID_AT) + with ( + _v1_cache() as write, + patch.object(live_rules.governance, "load_market_governance_states", return_value=[state]), + ): + result = evaluate_v1(_client(), Chain.MAINNET, V1_CONFIG, [_event("SubmitCap", {"id": MARKET, "cap": "1"})]) + + (alert,) = result.alerts + assert alert.severity == AlertSeverity.MEDIUM and "➕ Adding new market [WETH/USDC (86.00%)]" in alert.body + assert result.reasons == {"1_100_0": REASON_ALERTED} + assert result.header.startswith("[Yearn USDC](") + write.assert_not_called() + + +def test_v1_removal_of_a_supplied_market_is_a_standalone_loss() -> None: + state = _state(removable_at=VALID_AT, supply_shares=10**12, supply_assets=10**12, vault_total_assets=10**13) + with _v1_cache(), patch.object(live_rules.governance, "load_market_governance_states", return_value=[state]): + result = evaluate_v1(_client(), Chain.MAINNET, V1_CONFIG, [_event("SubmitMarketRemoval", {"id": MARKET})]) + + (alert,) = result.alerts + assert alert.standalone and alert.severity == AlertSeverity.HIGH + (severity, message), *_ = render_vault_alerts(result.header, result.alerts) + assert message.startswith(LOSS_PREFIX) and severity == AlertSeverity.HIGH + + +def test_v1_operation_no_longer_pending_or_already_alerted_is_not_new() -> None: + revoked = _state() # pendingCap / removableAt read back as 0 + already = _state(pending_cap=5, pending_cap_timestamp=VALID_AT) + for state, cache in ((revoked, {}), (already, {(MARKET, "pending_cap"): VALID_AT})): + with ( + _v1_cache(cache), + patch.object(live_rules.governance, "load_market_governance_states", return_value=[state]), + ): + result = evaluate_v1(_client(), Chain.MAINNET, V1_CONFIG, [_event("SubmitCap", {"id": MARKET})]) + assert result.alerts == [] and result.reasons == {"1_100_0": REASON_NOT_NEW} + + +def test_v1_cap_and_removal_on_one_market_are_checked_once() -> None: + state = _state(pending_cap=5, pending_cap_timestamp=VALID_AT, removable_at=VALID_AT) + events = [_event("SubmitCap", {"id": MARKET}), _event("SubmitMarketRemoval", {"id": MARKET}, log=1)] + with ( + _v1_cache(), + patch.object(live_rules.governance, "load_market_governance_states", return_value=[state]) as load, + ): + result = evaluate_v1(_client(), Chain.MAINNET, V1_CONFIG, events) + + assert len(load.call_args.args[1]) == 1 # one market read + assert len(result.alerts) == 2 # one cap section, one removal section: not doubled + assert set(result.reasons.values()) == {REASON_ALERTED} + + +def test_rules_wait_for_an_rpc_behind_the_event_block() -> None: + with pytest.raises(RpcBehindError): + evaluate_v1(_client(head=99), Chain.MAINNET, V1_CONFIG, [_event("SubmitCap", {"id": MARKET}, block=100)]) + + +# ── V2 rules ─────────────────────────────────────────────────────────── + + +def _snapshot(adapter_assets: int = 0, owner: str = NEW_OWNER) -> V2GovernanceSnapshot: + return V2GovernanceSnapshot( + name="Yearn USDC", + address=V2_VAULT, + chain=Chain.MAINNET, + owner=str(Web3.to_checksum_address(owner)), + curator=str(Web3.to_checksum_address("0x" + "cc" * 20)), + sentinels=[], + allocators=[], + adapters=[str(Web3.to_checksum_address(ADAPTER))], + asset_symbol="USDC", + asset_decimals=6, + total_assets=4_000_000 * 10**6, + adapter_assets={str(Web3.to_checksum_address(ADAPTER)): adapter_assets}, + ) + + +def _v2_client(executable_at: dict[bytes, int]) -> MagicMock: + client = _client() + contract = MagicMock() + contract.functions.executableAt.side_effect = lambda data: MagicMock( + call=lambda **_: executable_at.get(bytes(data), 0) + ) + client.eth.contract.return_value = contract + return client + + +@contextmanager +def _v2_cache(values: dict[str, str] | None = None) -> Iterator[MagicMock]: + values = values or {} + write = MagicMock() + with ( + patch( + "protocols.morpho.governance_v2.get_last_value_for_key_from_file", + side_effect=lambda _f, key: values.get(key, 0), + ), + patch("protocols.morpho.governance_v2.write_last_value_to_file", write), + ): + yield write + + +def _submit(data: str, log: int = 0) -> GovernanceEvent: + return _event("Submit", {"selector": data[:10], "data": data, "executableAt": str(VALID_AT)}, log=log, + vault=V2_VAULT, version=2) # fmt: skip + + +REMOVE = _calldata("removeAdapter(address)", ["address"], [ADAPTER]) +ABDICATE = _calldata("abdicate(bytes4)", ["bytes4"], [selector_of("setIsAllocator(address,bool)")]) +ADD = _calldata("addAdapter(address)", ["address"], ["0x" + "dd" * 20]) + + +def _bytes(data: str) -> bytes: + return bytes.fromhex(data[2:]) + + +def test_v2_remove_funded_adapter_is_a_standalone_loss_confirmed_on_chain() -> None: + client = _v2_client({_bytes(REMOVE): VALID_AT}) + with _v2_cache() as write, patch.object(live_rules, "load_v2_snapshot", return_value=_snapshot(10**12)): + result = evaluate_v2(client, Chain.MAINNET, V2_CONFIG, [_submit(REMOVE)]) + + (alert,) = result.alerts + assert alert.standalone and alert.severity == AlertSeverity.HIGH + assert "removeAdapter" in alert.body and "1.00M USDC" in alert.body + assert result.reasons == {"1_100_0": REASON_ALERTED} + write.assert_not_called() + + +def test_v2_new_submissions_are_grouped_like_the_daily_job() -> None: + client = _v2_client({_bytes(ABDICATE): VALID_AT, _bytes(ADD): VALID_AT}) + with _v2_cache(), patch.object(live_rules, "load_v2_snapshot", return_value=_snapshot()): + result = evaluate_v2(client, Chain.MAINNET, V2_CONFIG, [_submit(ABDICATE), _submit(ADD, log=1)]) + + (alert,) = result.alerts + assert alert.severity == AlertSeverity.MEDIUM and alert.body.startswith("📥 Submitted 2 operations:") + assert set(result.reasons.values()) == {REASON_ALERTED} + + +def test_v2_submit_already_accepted_or_revoked_is_not_new() -> None: + client = _v2_client({}) # executableAt(data) == 0 + with _v2_cache(), patch.object(live_rules, "load_v2_snapshot", return_value=_snapshot(10**12)): + result = evaluate_v2(client, Chain.MAINNET, V2_CONFIG, [_submit(REMOVE)]) + assert result.alerts == [] and result.reasons == {"1_100_0": REASON_NOT_NEW} + + +def test_v2_owner_change_uses_the_daily_role_cache_and_accept_is_left_to_the_daily_job() -> None: + old_owner = "0x" + "ee" * 20 + events = [ + _event("SetOwner", {"newOwner": NEW_OWNER}, vault=V2_VAULT, version=2), + _event("Accept", {"selector": ADD[:10], "data": ADD}, log=1, vault=V2_VAULT, version=2), + ] + with ( + _v2_cache({morpho_key(V2_VAULT.lower(), "owner", ROLE_TYPE): old_owner}), + patch.object(live_rules, "load_v2_snapshot", return_value=_snapshot()), + ): + result = evaluate_v2(_v2_client({}), Chain.MAINNET, V2_CONFIG, events) + + (alert,) = result.alerts + assert alert.severity == AlertSeverity.HIGH and "Owner changed" in alert.body + assert result.reasons == {"1_100_0": REASON_ALERTED, "1_100_1": REASON_RESOLUTION} + + +def test_load_v2_snapshot_reads_roles_adapters_and_values_on_chain() -> None: + client = _client() + vault = MagicMock() + vault.functions.owner.return_value.call.return_value = NEW_OWNER + vault.functions.curator.return_value.call.return_value = "0x" + "cc" * 20 + vault.functions.asset.return_value.call.return_value = "0x" + "a0" * 20 + vault.functions.adaptersLength.return_value.call.return_value = 1 + vault.functions.adapters.return_value.call.return_value = ADAPTER + vault.functions.totalAssets.return_value.call.side_effect = ValueError("execution reverted") + adapter = MagicMock() + adapter.functions.realAssets.return_value.call.return_value = 7 + client.eth.contract.side_effect = lambda address, abi: adapter if address.lower() == ADAPTER else vault + + with patch.object(live_rules, "fetch_erc20_metadata", return_value=MagicMock(symbol="USDC", decimals=6)): + snapshot = live_rules.load_v2_snapshot(client, Chain.MAINNET, V2_CONFIG, 1_000) + + checksummed = str(Web3.to_checksum_address(ADAPTER)) + assert snapshot.adapters == [checksummed] and snapshot.adapter_assets == {checksummed: 7} + assert snapshot.owner == Web3.to_checksum_address(NEW_OWNER) and snapshot.total_assets is None + assert (snapshot.asset_symbol, snapshot.asset_decimals) == ("USDC", 6) + + +# ── consumer: decisions, history, retries ────────────────────────────── + + +class FakeEvaluator: + """Stands in for the on-chain rules: one MEDIUM section per event, or a scripted failure.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, list[str]]] = [] + self.fail: list[Exception] = [] + + def __call__(self, chain: Chain, vault: VaultConfig, version: int, events: list[GovernanceEvent]): + self.calls.append((vault.address, [e.event_id for e in events])) + if self.fail: + raise self.fail.pop(0) + return VaultEvaluation( + header=f"[{vault.name}] on {chain.name}", + alerts=[VaultAlert(AlertSeverity.MEDIUM, f"📥 Submitted {e.name}") for e in events], + reasons={e.event_id: REASON_ALERTED for e in events}, + ) + + +def _store(tmp_path: Path) -> LiveStore: + return LiveStore(EventInbox(MORPHO_GOVERNANCE_STREAM, str(tmp_path / "morpho.db"))) + + +def _observe(store: LiveStore, *rows: dict[str, Any], source: str = SOURCE_WS, at: float = 1_700_000_100.0) -> None: + store.inbox.observe(list(rows), source, at) + + +def _consumer(store: LiveStore, evaluate: Any, clock: list[float], **config: Any) -> LiveGovernanceConsumer: + return LiveGovernanceConsumer( + store, LiveConfig(**config), evaluate=evaluate, send=MagicMock(), now=lambda: clock[0] + ) + + +def test_history_before_the_baseline_is_never_evaluated(tmp_path: Path) -> None: + store = _store(tmp_path) + evaluate = FakeEvaluator() + clock = [1_700_000_000.0] + consumer = _consumer(store, evaluate, clock) + consumer.load_baseline() + _observe(store, make_morpho_event(1, 10, ts=1_699_999_000), make_morpho_event(1, 11, ts=1_700_000_050)) + + assert consumer.process_once() == 2 + + assert evaluate.calls == [(V1_VAULT, ["1_11_0"])] + outcomes = {row["event_id"]: row["outcome"] for row in store.decisions_since(0)} + assert outcomes == {"1_10_0": OUTCOME_HISTORICAL, "1_11_0": OUTCOME_ALERTED} + assert store.inbox.status()[BASELINE_STATUS_KEY][0] == 1_700_000_000.0 + + +def test_observe_mode_records_grouped_requests_and_sends_nothing(tmp_path: Path) -> None: + store = _store(tmp_path) + evaluate = FakeEvaluator() + send = MagicMock() + consumer = LiveGovernanceConsumer(store, LiveConfig(), evaluate=evaluate, send=send, now=lambda: 1_000.0) + other_vault = OTHER_VAULT + _observe( + store, make_morpho_event(1, 10), make_morpho_event(1, 10, 1), make_morpho_event(1, 12, address=other_vault) + ) + + consumer.process_once() + + # per vault, in chain order + assert evaluate.calls == [(V1_VAULT, ["1_10_0", "1_10_1"]), (OTHER_VAULT, ["1_12_0"])] + alerts = store.alerts_since(0) + assert [a["status"] for a in alerts] == [STATUS_OBSERVED, STATUS_OBSERVED] # one message per vault + assert json.loads(alerts[0]["event_ids"]) == ["1_10_0", "1_10_1"] + assert store.pending_alerts() == [] and store.undecided() == [] + send.assert_not_called() + + +def test_unconfigured_vault_and_an_empty_evaluation_record_no_alert(tmp_path: Path) -> None: + store = _store(tmp_path) + + def nothing(chain, vault, version, events): # noqa: ANN001, ANN202 + return VaultEvaluation(header="h", reasons={e.event_id: REASON_NOT_NEW for e in events}) + + _observe(store, make_morpho_event(1, 10), make_morpho_event(1, 11, address="0x" + "99" * 20)) + _consumer(store, nothing, [1_000.0]).process_once() + + reasons = {row["event_id"]: (row["outcome"], row["reason"]) for row in store.decisions_since(0)} + assert reasons == { + "1_10_0": (OUTCOME_NO_ALERT, REASON_NOT_NEW), + "1_11_0": (OUTCOME_NO_ALERT, "vault not in config.py"), + } + assert store.alerts_since(0) == [] + + +def test_rpc_lag_and_failures_retry_with_backoff_without_blocking_other_vaults(tmp_path: Path) -> None: + store = _store(tmp_path) + evaluate = FakeEvaluator() + evaluate.fail = [RpcBehindError("head behind"), RuntimeError("rpc down")] + clock = [1_000.0] + consumer = _consumer(store, evaluate, clock, retry_initial=5.0) + _observe(store, make_morpho_event(1, 10)) + + consumer.process_once() # RPC behind + consumer.process_once() # still inside the retry delay: not called again + clock[0] += 6 + consumer.process_once() # fails + clock[0] += 6 + consumer.process_once() # inside the doubled delay (10s) + clock[0] += 5 + consumer.process_once() # succeeds + + assert len(evaluate.calls) == 3 + assert [row["outcome"] for row in store.decisions_since(0)] == [OUTCOME_ALERTED] + + +def test_restart_resumes_undecided_events_and_keeps_the_baseline(tmp_path: Path) -> None: + store = _store(tmp_path) + clock = [1_700_000_000.0] + _consumer(store, FakeEvaluator(), clock).process_once() + _observe(store, make_morpho_event(1, 10, ts=1_700_000_010)) + + clock[0] = 1_800_000_000.0 # restarted much later: the event is still after the stored baseline + restarted = _consumer(_store(tmp_path), FakeEvaluator(), clock) + restarted.process_once() + + assert restarted.baseline == 1_700_000_000.0 + assert [row["outcome"] for row in store.decisions_since(0)] == [OUTCOME_ALERTED] + + +def test_a_reorg_rewrite_is_evaluated_again(tmp_path: Path) -> None: + store = _store(tmp_path) + evaluate = FakeEvaluator() + consumer = _consumer(store, evaluate, [1_000.0]) + _observe(store, make_morpho_event(1, 10)) + consumer.process_once() + consumer.process_once() # decided: nothing to do + _observe(store, make_morpho_event(1, 10, blockHash="0x" + "dd" * 32), source=SOURCE_HTTP) + consumer.process_once() + + assert [ids for _, ids in evaluate.calls] == [["1_10_0"], ["1_10_0"]] + (decision,) = store.decisions_since(0) + assert decision["revision"] == 1 + + +# ── consumer: shadow delivery ────────────────────────────────────────── + + +def test_shadow_requests_are_delivered_in_order_and_failures_retried(tmp_path: Path) -> None: + store = _store(tmp_path) + sent: list[int] = [] + attempts = {"n": 0} + + def send(request: AlertRequest) -> None: + attempts["n"] += 1 + if attempts["n"] == 1: + raise RuntimeError("telegram 502") + sent.append(request.id) + + consumer = LiveGovernanceConsumer( + store, + LiveConfig(mode=MODE_SHADOW, poll_interval=0.02, delivery_interval=0.01, retry_initial=0.05), + evaluate=FakeEvaluator(), + send=send, + now=time.time, + ) + other_vault = OTHER_VAULT + _observe( + store, + make_morpho_event(1, 10, ts=int(time.time())), + make_morpho_event(1, 11, ts=int(time.time()), address=other_vault), + ) + + async def scenario() -> None: + task = asyncio.create_task(consumer.run()) + deadline = time.monotonic() + 5 + while len(sent) < 2 and time.monotonic() < deadline: + await asyncio.sleep(0.02) + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + asyncio.run(scenario()) + + alerts = store.alerts_since(0) + assert sent == [alerts[0]["id"], alerts[1]["id"]] # the failed first request still went first + assert [a["status"] for a in alerts] == [STATUS_SENT, STATUS_SENT] + assert alerts[0]["attempts"] == 2 and alerts[1]["attempts"] == 1 + + +def test_shadow_requires_the_errors_chat(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + assert build_consumer(_store(tmp_path), MODE_SHADOW).config.mode == "observe" # conftest strips TELEGRAM_* + request = AlertRequest(1, 1, V1_VAULT, MODE_SHADOW, "HIGH", "🚨 POTENTIAL VAULT LOSS — x", ["1_10_0"], 0) + with pytest.raises(ShadowChannelMissing): + send_shadow(request) # conftest strips TELEGRAM_* + + monkeypatch.setenv("TELEGRAM_CHAT_ID_ERRORS", "-100123") + with patch("protocols.morpho.live_governance.send_telegram_message", return_value=True) as send: + send_shadow(request) + text, channel, _silent = send.call_args.args + assert channel == "errors" and text.startswith(f"[{SHADOW_LABEL}] shadow → morpho, HIGH, events 1_10_0") + assert send.call_args.kwargs["plain_text"] and send.call_args.kwargs["origin_protocol"] == "yearn-internal" + + # The errors chat id is set but the bot token is not: Telegram skips it, so it is not "sent". + with pytest.raises(NotDelivered): + send_shadow(request) + + +def test_shadow_text_bounds_its_header_and_fits_telegram() -> None: + ids = [f"1_{n}_0" for n in range(500)] + body = "\n".join("x" * 100 for _ in range(60)) # longer than one Telegram message + texts = shadow_texts(AlertRequest(1, 1, V1_VAULT, MODE_SHADOW, "MEDIUM", body, ids, 0)) + + assert len(texts) > 1 and all(len(t) <= MAX_MESSAGE_LENGTH for t in texts) + assert "events 1_0_0, 1_1_0, 1_2_0, 1_3_0, 1_4_0 (+495 more)" in texts[0] + assert sum(t.count("x" * 100) for t in texts) == 60 # nothing dropped + + +def test_production_does_not_commit_when_telegram_skips_the_alert() -> None: + with patch("protocols.morpho.live_governance.send_alert", return_value=False), pytest.raises(NotDelivered): + publish_to_morpho(AlertSeverity.HIGH, "🚨 x") + + +def test_live_health_and_summary(tmp_path: Path) -> None: + store = _store(tmp_path) + clock = [1_000.0] + consumer = LiveGovernanceConsumer( + store, LiveConfig(mode=MODE_SHADOW), evaluate=FakeEvaluator(), now=lambda: clock[0] + ) + consumer.load_baseline() + _observe(store, make_morpho_event(1, 10, ts=1_000), at=1_005.0) + clock[0] = 1_010.0 + consumer.process_once() + + assert live_health(store, 1_100.0) == [] + assert live_health(store, 2_000.0) == ["live: 1 shadow alert(s) undelivered, oldest 990s"] + summary = live_summary(store, 1, 1_100.0) + assert "decisions: 1 {'alerted': 1}" in summary and f"'{STATUS_PENDING}': 1" in summary + assert "block → decision: p50 10.0s, p95 10.0s" in summary + + +# ── worker integration ───────────────────────────────────────────────── + + +def test_worker_wakes_the_consumer_after_a_websocket_ingest(tmp_path: Path) -> None: + async def scenario() -> None: + envio = FakeEnvio() + envio.set_head(1, 1) + server = FakeWSServer(envio) + await server.start() + store = _store(tmp_path) + evaluate = FakeEvaluator() + # A long poll interval: only the ingest wake-up can make the decision this fast. + consumer = LiveGovernanceConsumer(store, LiveConfig(poll_interval=60), evaluate=evaluate) + config = WorkerConfig( + reconcile_interval=60, + heartbeat_interval=0.05, + backoff_initial=0.05, + reconcile=ReconcileConfig(page_size=20, overlap_blocks={1: 5}, max_block_span={1: 50}), + ) + worker = Worker(envio.post, server.url, store.inbox, {1: {V1_VAULT.lower()}}, config, consumer=consumer) + task = asyncio.create_task(worker.run()) + try: + deadline = time.monotonic() + 5 + while not (worker.ws_connected and consumer.baseline) and time.monotonic() < deadline: + await asyncio.sleep(0.02) + await asyncio.sleep(0.1) # let the consumer finish its first (empty) pass + envio.add(make_morpho_event(1, 10, ts=int(time.time()) + 1)) + while not store.decisions_since(0) and time.monotonic() < deadline: + await asyncio.sleep(0.02) + assert worker.is_alive() + finally: + worker.stop() + await asyncio.wait_for(task, 5) + await server.stop() + assert [row["outcome"] for row in store.decisions_since(0)] == [OUTCOME_ALERTED] + + with patch("utils.subscriber.worker.notify"): + asyncio.run(scenario()) + + +def test_a_dead_consumer_makes_the_worker_unhealthy(tmp_path: Path) -> None: + class Crashing: + async def run(self) -> None: + raise RuntimeError("bug") + + def wake(self) -> None: + pass + + async def scenario() -> None: + store = _store(tmp_path) + config = WorkerConfig(transport="http", reconcile_interval=60, heartbeat_interval=0.05) + worker = Worker(FakeEnvio().post, None, store.inbox, {1: {V1_VAULT.lower()}}, config, consumer=Crashing()) + task = asyncio.create_task(worker.run()) + try: + deadline = time.monotonic() + 5 + while worker.is_alive() and time.monotonic() < deadline: + await asyncio.sleep(0.02) + assert not worker.is_alive() + finally: + worker.stop() + await asyncio.wait_for(task, 5) + + with patch("utils.subscriber.worker.notify"): + asyncio.run(scenario()) + + +# ── production mode ──────────────────────────────────────────────────── + + +class CommittingEvaluator(FakeEvaluator): + """FakeEvaluator whose evaluations record when their cache writes are committed.""" + + def __init__(self, log: list[str]) -> None: + super().__init__() + self.log = log + + def __call__(self, chain: Chain, vault: VaultConfig, version: int, events: list[GovernanceEvent]): + self.log.append("evaluate") + result = super().__call__(chain, vault, version, events) + result.commit = lambda: self.log.append("commit") + return result + + +@contextmanager +def _recording_lock(log: list[str]) -> Iterator[MagicMock]: + @contextmanager + def lock(address: str) -> Iterator[None]: + log.append(f"lock {address}") + yield + log.append("unlock") + + with patch("protocols.morpho.live_governance.governance_lock", side_effect=lock) as patched: + yield patched + + +def test_production_sends_then_commits_under_the_vault_lock(tmp_path: Path) -> None: + store = _store(tmp_path) + log: list[str] = [] + + def publish(severity: AlertSeverity, message: str) -> None: + log.append(f"publish {severity.value}") + + consumer = LiveGovernanceConsumer( + store, LiveConfig(mode=MODE_PRODUCTION), evaluate=CommittingEvaluator(log), publish=publish, now=lambda: 1.0 + ) + _observe(store, make_morpho_event(1, 10)) + with _recording_lock(log): + consumer.process_once() + + assert log == [f"lock {V1_VAULT}", "evaluate", "publish MEDIUM", "commit", "unlock"] + (alert,) = store.alerts_since(0) + assert alert["status"] == STATUS_SENT and alert["mode"] == MODE_PRODUCTION + assert [row["outcome"] for row in store.decisions_since(0)] == [OUTCOME_ALERTED] + + +def test_production_send_failure_commits_nothing_and_retries_fresh(tmp_path: Path) -> None: + store = _store(tmp_path) + log: list[str] = [] + failures = ["telegram 502"] + + def publish(severity: AlertSeverity, message: str) -> None: + if failures: + raise RuntimeError(failures.pop()) + log.append("publish") + + clock = [1.0] + evaluate = CommittingEvaluator(log) + consumer = LiveGovernanceConsumer( + store, LiveConfig(mode=MODE_PRODUCTION, retry_initial=5.0), evaluate=evaluate, publish=publish, + now=lambda: clock[0], + ) # fmt: skip + _observe(store, make_morpho_event(1, 10)) + with _recording_lock(log): + consumer.process_once() # send fails + assert "commit" not in log and store.decisions_since(0) == [] and store.alerts_since(0) == [] + clock[0] += 6 + consumer.process_once() # re-evaluated against fresh state, sent, committed + + assert log.count("evaluate") == 2 and log.count("commit") == 1 + assert [row["outcome"] for row in store.decisions_since(0)] == [OUTCOME_ALERTED] + + +def test_daily_v2_job_takes_the_same_vault_lock() -> None: + entered: list[str] = [] + + @contextmanager + def lock(address: str) -> Iterator[None]: + entered.append(address) + yield + + with ( + patch("protocols.morpho.governance_v2.governance_lock", side_effect=lock), + _v2_cache(), + patch("protocols.morpho.governance_v2.send_vault_alerts"), + ): + governance_v2.diff_and_alert(_snapshot()) + + assert entered == [V2_VAULT] + + +def test_file_lock_excludes_a_second_holder_until_released() -> None: + with file_lock("test_vault"): + with pytest.raises(LockTimeout), file_lock("test_vault", timeout=0.05, poll_interval=0.01): + pass + with file_lock("test_vault", timeout=0.05): + pass + + +# ── review fixes ─────────────────────────────────────────────────────── + + +def test_a_failing_adapter_valuation_retries_instead_of_reading_as_empty() -> None: + client = _client() + vault = MagicMock() + vault.functions.owner.return_value.call.return_value = NEW_OWNER + vault.functions.curator.return_value.call.return_value = NEW_OWNER + vault.functions.asset.return_value.call.return_value = "0x" + "a0" * 20 + vault.functions.adaptersLength.return_value.call.return_value = 1 + vault.functions.adapters.return_value.call.return_value = ADAPTER + adapter = MagicMock() + adapter.functions.realAssets.return_value.call.side_effect = TimeoutError("rpc timeout") + client.eth.contract.side_effect = lambda address, abi: adapter if address.lower() == ADAPTER else vault + + with pytest.raises(TimeoutError): + live_rules.load_v2_snapshot(client, Chain.MAINNET, V2_CONFIG, 1_000) + + +def test_backed_off_vaults_do_not_crowd_other_vaults_out_of_the_batch(tmp_path: Path) -> None: + store = _store(tmp_path) + clock = [1_000.0] + calls: list[str] = [] + + def evaluate(chain, vault, version, events): # noqa: ANN001, ANN202 + calls.append(vault.address) + if vault.address == V1_VAULT: + raise RuntimeError("rpc down") + return FakeEvaluator()(chain, vault, version, events) + + consumer = _consumer(store, evaluate, clock, batch_limit=3, retry_initial=60.0) + _observe(store, *(make_morpho_event(1, block) for block in range(10, 15))) # 5 events of the failing vault + _observe(store, make_morpho_event(1, 20, address=OTHER_VAULT)) + + consumer.process_once() # the first 3 rows are all the failing vault's + consumer.process_once() # it is backed off: the batch reaches the other vault + + assert calls == [V1_VAULT, OTHER_VAULT] + assert [row["event_id"] for row in store.decisions_since(0)] == ["1_20_0"] + + +def test_backoff_survives_a_very_long_outage(tmp_path: Path) -> None: + store = _store(tmp_path) + evaluate = FakeEvaluator() + evaluate.fail = [RuntimeError("down")] + consumer = _consumer(store, evaluate, [1_000.0], retry_max=300.0) + consumer._retry[(1, V1_VAULT.lower())] = (5_000, 0.0) # thousands of failures already + _observe(store, make_morpho_event(1, 10)) + + consumer.process_once() + + assert consumer._retry[(1, V1_VAULT.lower())] == (5_001, 1_300.0) + + +def test_pending_shadow_requests_are_invalidated_when_their_event_is_reorged(tmp_path: Path) -> None: + store = _store(tmp_path) + consumer = _consumer(store, FakeEvaluator(), [1_000.0], mode=MODE_SHADOW) + _observe(store, make_morpho_event(1, 10), make_morpho_event(1, 11, address=OTHER_VAULT)) + consumer.process_once() + assert len(store.pending_alerts()) == 2 + + store.inbox.mark_missing(1, 10, 10, set(), SOURCE_HTTP, 1_001.0) # reorg dropped the first event + _observe(store, make_morpho_event(1, 11, address=OTHER_VAULT, blockHash="0x" + "dd" * 32), source=SOURCE_HTTP) + + assert store.pending_alerts() == [] + assert [row["status"] for row in store.alerts_since(0)] == [STATUS_INVALIDATED, STATUS_INVALIDATED] + consumer.process_once() # the rewritten event is evaluated again and makes its own request + (replacement,) = store.pending_alerts() + assert replacement.event_ids == ["1_11_0"] + + +def test_subscriber_reads_the_daily_governance_cache_namespace() -> None: + import os # noqa: PLC0415 + import subprocess # noqa: PLC0415 + import sys # noqa: PLC0415 + + env = {k: v for k, v in os.environ.items() if k != "MORPHO_FILENAME"} + code = ( + "import os, protocols.timelock.event_subscriber; from utils import cache;" + "print(os.path.basename(cache.morpho_filename))" + ) + out = subprocess.run([sys.executable, "-c", code], env=env, capture_output=True, text=True, check=True) + + assert out.stdout.strip().splitlines()[-1] == daily_morpho_filename() == "cache-id-daily.txt" + + +def test_check_daily_governance_cache_rejects_another_namespace() -> None: + with patch("utils.cache.morpho_filename", "/srv/cache/cache-id.txt"), pytest.raises(RuntimeError, match="daily"): + check_daily_governance_cache() diff --git a/tests/test_morpho_v2_governance.py b/tests/test_morpho_v2_governance.py index 3a86bba0..14a8851c 100644 --- a/tests/test_morpho_v2_governance.py +++ b/tests/test_morpho_v2_governance.py @@ -85,7 +85,7 @@ def test_resolved_pending_alert_without_cached_function_keeps_hash_only_message( sent: list[Any] = [] with patch("protocols.morpho._alerts.send_alert", side_effect=sent.append): - governance_v2.send_vault_alerts(governance_v2._vault_header(snapshot), diff.alerts, governance_v2.PROTOCOL) + governance_v2.send_vault_alerts(governance_v2.vault_header(snapshot), diff.alerts, governance_v2.PROTOCOL) self.assertEqual(len(sent), 1) message = sent[0].message @@ -275,11 +275,11 @@ def test_oversized_group_splits_into_numbered_parts(self) -> None: diff = governance_v2.VaultDiff() snapshot = _snapshot(pending) for pc in pending: - governance_v2._alert_pending_new(snapshot, [(pc, "increaseTimelock(setSendAssetsGate → 604800s)")], diff) + governance_v2.alert_pending_new(snapshot, [(pc, "increaseTimelock(setSendAssetsGate → 604800s)")], diff) sent: list[Any] = [] with patch("protocols.morpho._alerts.send_alert", side_effect=sent.append): - governance_v2.send_vault_alerts(governance_v2._vault_header(snapshot), diff.alerts, governance_v2.PROTOCOL) + governance_v2.send_vault_alerts(governance_v2.vault_header(snapshot), diff.alerts, governance_v2.PROTOCOL) self.assertGreater(len(sent), 1, "oversized group should split into multiple messages") for index, alert in enumerate(sent, start=1): @@ -308,14 +308,14 @@ def test_oversized_pending_section_splits_without_losing_operations(self) -> Non for i, pc in enumerate(pending): label = f"increaseTimelock(setSendAssetsGate → {604800 + i}s)" operations.append((pc, label)) - governance_v2._alert_pending_new(snapshot, operations, diff) + governance_v2.alert_pending_new(snapshot, operations, diff) self.assertEqual(len(diff.alerts), 1, "the regression requires one oversized section") self.assertGreater(len(diff.alerts[0].body), MAX_MESSAGE_LENGTH) sent: list[Any] = [] with patch("protocols.morpho._alerts.send_alert", side_effect=sent.append): - governance_v2.send_vault_alerts(governance_v2._vault_header(snapshot), diff.alerts, governance_v2.PROTOCOL) + governance_v2.send_vault_alerts(governance_v2.vault_header(snapshot), diff.alerts, governance_v2.PROTOCOL) self.assertGreater(len(sent), 1) for alert in sent: diff --git a/tests/test_timelock_subscriber_reconcile.py b/tests/test_timelock_subscriber_reconcile.py index 36961b45..56c63d3b 100644 --- a/tests/test_timelock_subscriber_reconcile.py +++ b/tests/test_timelock_subscriber_reconcile.py @@ -4,12 +4,13 @@ import sqlite3 import threading from pathlib import Path +from typing import Any import pytest from protocols.timelock.event_stream import TIMELOCK_STREAM from tests.timelock_subscriber_fakes import OTHER_TIMELOCK, TIMELOCK, FakeEnvio, make_event -from utils.subscriber.envio import EnvioQueryError, fetch_range, range_query +from utils.subscriber.envio import EnvioQueryError, entity_rows, fetch_range, range_query from utils.subscriber.inbox import SCHEMA_VERSION, SOURCE_HTTP, SOURCE_WS, EventInbox, event_fingerprint from utils.subscriber.reconcile import ReconcileConfig, build_scope, in_scope, reconcile_all @@ -396,3 +397,75 @@ def test_widened_selection_does_not_hide_a_real_change(tmp_path: Path) -> None: result = inbox.observe([make_event(1, 10, transactionHash="0x" + "ee" * 32)], SOURCE_HTTP, 20.0) assert result.changed == 1 assert [c["kind"] for c in inbox.changes_since(0)] == ["changed"] + + +# ── malformed responses fail closed ──────────────────────────────────── + +# A page reconciliation must not mistake for an empty range (issue #437 review): each +# would otherwise mark the range's events removed and move the checkpoint past them. +MALFORMED_PAGES = [ + {}, + {"data": None}, + {"data": {}}, + {"data": {"TimelockEvent": None}}, + {"data": {"TimelockEvent": "oops"}}, + {"data": {"TimelockEvent": [None]}}, + [], +] + + +def _scenario(inbox: EventInbox) -> FakeEnvio: + """An active event at block 12, checkpoint 15, overlap 5, indexed head 20 (chain 1 only).""" + envio = FakeEnvio() + envio.set_head(1, 15) + reconcile_all(envio.post, inbox, {1: {TIMELOCK}}, CONFIG) + inbox.observe([make_event(1, 12)], SOURCE_WS, 1.0) + envio.add(make_event(1, 12), advance_head=False) + envio.set_head(1, 20) + return envio + + +@pytest.mark.parametrize("page", MALFORMED_PAGES, ids=repr) +def test_malformed_range_page_keeps_events_and_checkpoint(inbox: EventInbox, page: object) -> None: + envio = _scenario(inbox) + + def post(query: str, variables: dict) -> Any: # deliberately not a well-formed response + return envio.post(query, variables) if "chain_metadata" in query else page + + report = reconcile_all(post, inbox, {1: {TIMELOCK}}, CONFIG) + + assert not report.ok and "GraphQL response" in (report.chains[0].error or "") + (row,) = inbox.events_since(0) + assert row["status"] == "active" + assert _checkpoint(inbox, 1) == 15 + assert inbox.changes_since(0) == [] + + +def test_explicit_empty_page_still_completes_the_range(inbox: EventInbox) -> None: + envio = _scenario(inbox) + envio.remove("1_12_0") # the indexer really dropped it: [] is a genuine empty range + + assert reconcile_all(envio.post, inbox, {1: {TIMELOCK}}, CONFIG).ok + + (row,) = inbox.events_since(0) + assert row["status"] == "removed" and _checkpoint(inbox, 1) == 20 + + +@pytest.mark.parametrize("payload", [{}, {"data": None}, {"data": {"chain_metadata": None}}], ids=repr) +def test_malformed_chain_metadata_fails_the_cycle(inbox: EventInbox, payload: object) -> None: + _scenario(inbox) + + def post(query: str, variables: dict) -> Any: # deliberately not a well-formed response + return payload + + report = reconcile_all(post, inbox, {1: {TIMELOCK}}, CONFIG) + + assert report.error is not None and "GraphQL response" in report.error + assert report.chains == [] and _checkpoint(inbox, 1) == 15 + + +def test_entity_rows_accepts_only_a_list_of_rows() -> None: + assert entity_rows({"data": {"TimelockEvent": []}}, "TimelockEvent") == [] + assert entity_rows({"data": {"TimelockEvent": [{"id": "1"}]}}, "TimelockEvent") == [{"id": "1"}] + with pytest.raises(EnvioQueryError, match="GraphQL errors"): + entity_rows({"errors": [{"message": "x"}], "data": {"TimelockEvent": []}}, "TimelockEvent") diff --git a/tests/test_timelock_subscriber_worker.py b/tests/test_timelock_subscriber_worker.py index b7d0f7a8..0ef1ce20 100644 --- a/tests/test_timelock_subscriber_worker.py +++ b/tests/test_timelock_subscriber_worker.py @@ -22,6 +22,7 @@ from tests.timelock_subscriber_fakes import TIMELOCK, FakeEnvio, FakeWSServer, make_event from utils.graphql_ws import GraphQLWSError, http_to_ws_url, subscribe from utils.sd_notify import notify +from utils.subscriber.envio import EnvioQueryError from utils.subscriber.inbox import SOURCE_HTTP, SOURCE_WS, EventInbox from utils.subscriber.reconcile import ReconcileConfig, reconcile_all from utils.subscriber.report import PilotReport, _recommend, build_report, format_report, percentile @@ -450,3 +451,13 @@ def test_report_ignores_closed_outage_before_the_window(tmp_path: Path) -> None: inbox.log("ws_connected", 10.0) report = build_report(inbox, hours=1, now=7_200.0) assert report.outages == [] and report.ws_downtime_seconds == 0.0 + + +@pytest.mark.parametrize("payload", [{}, {"data": None}, {"data": {"TimelockEvent": None}}], ids=repr) +def test_malformed_ws_snapshot_raises_instead_of_reading_as_empty(tmp_path: Path, payload: dict) -> None: + inbox = EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")) + worker = Worker(FakeEnvio().post, "ws://unused", inbox, SCOPE, _config()) + + with pytest.raises(EnvioQueryError, match="GraphQL response"): + asyncio.run(worker._ingest_snapshot(payload, 1.0)) + assert worker.ws_messages == 0 and _rows(inbox) == {} diff --git a/utils/alert.py b/utils/alert.py index 4fd9dc8f..14f4c819 100644 --- a/utils/alert.py +++ b/utils/alert.py @@ -112,7 +112,7 @@ def send_alert( *, silent: bool | None = None, plain_text: bool = False, -) -> None: +) -> bool: """Send an alert via Telegram with auto-emoji prefix and severity-based defaults. Args: @@ -121,6 +121,10 @@ def send_alert( silent: Override notification silencing. None uses severity default (LOW/MEDIUM=silent, HIGH/CRITICAL=loud). plain_text: If True, send without Markdown formatting. + + Returns: + True when Telegram accepted the message; False when it was skipped + (debug mode or missing credentials). See ``send_telegram_message``. """ emoji = _SEVERITY_EMOJI[alert.severity.value] message = f"{emoji} {alert.message}" @@ -128,7 +132,7 @@ def send_alert( if silent is None: silent = _SEVERITY_SILENT_DEFAULT[alert.severity.value] - send_telegram_message( + delivered = send_telegram_message( message, alert.channel or alert.protocol, silent, @@ -145,6 +149,7 @@ def send_alert( _alert_hook(alert) except Exception: logger.exception("Alert hook failed for %s alert", alert.severity.value) + return delivered # After AlertSeverity / Alert exist so utils.dispatch can import this module safely. diff --git a/utils/file_lock.py b/utils/file_lock.py new file mode 100644 index 00000000..9f26a609 --- /dev/null +++ b/utils/file_lock.py @@ -0,0 +1,48 @@ +"""Cross-process exclusive locks on files in ``CACHE_DIR``. + +Two processes that both check-then-write the same cache keys (a scheduled job +and a continuous worker) take the same named lock around the whole +read → alert → write sequence, so they cannot both see a change as new and +both alert it. ``flock`` locks are released by the kernel when the holder +exits, so a crash never leaves a stale lock. +""" + +import fcntl +import time +from collections.abc import Iterator +from contextlib import contextmanager + +from utils.paths import cache_path + + +class LockTimeout(Exception): + """The lock was still held by another process when the timeout ran out.""" + + +@contextmanager +def file_lock(name: str, timeout: float = 120.0, poll_interval: float = 0.1) -> Iterator[None]: + """Hold an exclusive lock on ``$CACHE_DIR/.lock`` for the ``with`` block. + + Args: + name: Lock name; callers that must exclude each other use the same name. + timeout: Seconds to wait for the lock. + poll_interval: Seconds between attempts while waiting. + + Raises: + LockTimeout: Another holder kept the lock for longer than ``timeout``. + """ + path = cache_path(f"{name}.lock") + with open(path, "a+") as handle: + deadline = time.monotonic() + timeout + while True: + try: + fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except BlockingIOError: + if time.monotonic() >= deadline: + raise LockTimeout(f"{path} still locked after {timeout:.0f}s") from None + time.sleep(poll_interval) + try: + yield + finally: + fcntl.flock(handle, fcntl.LOCK_UN) diff --git a/utils/subscriber/envio.py b/utils/subscriber/envio.py index 57a76239..2b3eb439 100644 --- a/utils/subscriber/envio.py +++ b/utils/subscriber/envio.py @@ -67,10 +67,29 @@ def post(query: str, variables: dict[str, Any]) -> dict[str, Any]: return post -def _data(payload: dict[str, Any]) -> dict[str, Any]: +def entity_rows(payload: Any, field: str) -> list[dict[str, Any]]: + """Return ``payload["data"][field]``, failing closed on anything but a list of rows. + + Reconciliation treats a returned page as the complete content of its block + range: rows missing from it are marked removed and the checkpoint moves past + it. So only an explicit list (``[]`` included) may count as a page. A missing + or null ``data``, a missing or null ``field``, or a non-object row is a broken + response, not an empty range, and raises before anything is changed. + + Raises: + EnvioQueryError: GraphQL ``errors``, or a response without the expected list. + """ + if not isinstance(payload, dict): + raise EnvioQueryError(f"GraphQL response is not an object: {str(payload)[:200]}") if payload.get("errors"): raise EnvioQueryError(f"GraphQL errors: {json.dumps(payload['errors'])[:500]}") - return payload.get("data") or {} + data = payload.get("data") + if not isinstance(data, dict): + raise EnvioQueryError(f"GraphQL response has no data object: {json.dumps(payload)[:200]}") + rows = data.get(field) + if not isinstance(rows, list) or not all(isinstance(row, dict) for row in rows): + raise EnvioQueryError(f"GraphQL response has no {field} list: {json.dumps(data)[:200]}") + return rows def address_variants(addresses: set[str]) -> list[str]: @@ -80,7 +99,7 @@ def address_variants(addresses: set[str]) -> list[str]: def fetch_chain_heads(post: GraphQLPost) -> dict[int, ChainHead]: """Fetch ``chain_metadata`` and key it by chain id.""" - rows = _data(post(CHAIN_METADATA_QUERY, {})).get("chain_metadata") or [] + rows = entity_rows(post(CHAIN_METADATA_QUERY, {}), "chain_metadata") heads: dict[int, ChainHead] = {} for row in rows: if row.get("latest_processed_block") is None: @@ -175,7 +194,7 @@ def fetch_range( rows: list[dict[str, Any]] = [] for page in range(max_pages): query = range_query(stream, chain_id, from_block, to_block, page_size, page * page_size, since_ts) - batch = _data(post(query, variables)).get(stream.entity) or [] + batch = entity_rows(post(query, variables), stream.entity) rows.extend(batch) if len(batch) < page_size: return rows, page + 1 diff --git a/utils/subscriber/worker.py b/utils/subscriber/worker.py index b7ec4e68..8105ad00 100644 --- a/utils/subscriber/worker.py +++ b/utils/subscriber/worker.py @@ -17,6 +17,11 @@ streams share the HTTP poster and a ``SharedChainHeads`` reader, and each keeps its own WebSocket connection, inbox, and checkpoints. +An optional ``StreamConsumer`` runs as one more loop of the worker, woken after +every ingest that stored new, changed, or restored rows. It reads the inbox +itself, so whatever it does (RPC reads, Telegram) never delays ingestion, and a +consumer that dies withholds the watchdog ping like any other loop. + The pilot sends no alerts. With an optional ``PilotNotices`` it posts diagnostic notices to the internal errors chat from a background queue (see ``notices.py``); without one, all output stays in journald and the inbox. @@ -29,12 +34,12 @@ from collections.abc import AsyncIterator, Callable from dataclasses import dataclass from time import monotonic -from typing import Any +from typing import Any, Protocol from utils.graphql_ws import GraphQLWSError, subscribe from utils.logger import get_logger from utils.sd_notify import notify -from utils.subscriber.envio import ChainHead, GraphQLPost, address_variants, live_subscription +from utils.subscriber.envio import ChainHead, GraphQLPost, address_variants, entity_rows, live_subscription from utils.subscriber.inbox import SOURCE_HTTP, SOURCE_WS, EventInbox from utils.subscriber.notices import PilotNotices from utils.subscriber.reconcile import ReconcileConfig, ReconcileReport, in_scope, reconcile_all @@ -45,6 +50,19 @@ SubscribeFn = Callable[..., AsyncIterator[dict[str, Any]]] + +class StreamConsumer(Protocol): + """Reads a stream's inbox on its own schedule (e.g. to decide alerts), never inside ingestion.""" + + async def run(self) -> None: + """Run until cancelled.""" + ... + + def wake(self) -> None: + """Called after an ingest stored new, changed, or restored rows.""" + ... + + # Doublings beyond this cannot matter (2**20 x any sane initial delay exceeds any # cap) and keep ``2 ** n`` from overflowing a float after ~1,000 failures. _MAX_BACKOFF_DOUBLINGS = 20 @@ -84,6 +102,7 @@ def __init__( now: Callable[[], float] = time.time, notices: PilotNotices | None = None, heads: Callable[[], dict[int, ChainHead]] | None = None, + consumer: StreamConsumer | None = None, ) -> None: if config.transport not in TRANSPORTS: raise ValueError(f"transport must be one of {TRANSPORTS}, got {config.transport!r}") @@ -98,6 +117,7 @@ def __init__( self.now = now self.notices = notices self.heads = heads + self.consumer = consumer self.stream = inbox.stream # False when ``run_workers`` owns signals and systemd notifications for the process. self.supervised = True @@ -140,6 +160,8 @@ async def run(self) -> None: if self.config.transport in ("both", "ws"): loops.append(asyncio.create_task(self.ws_loop(), name="ws")) loops.append(asyncio.create_task(self.heartbeat_loop(), name="heartbeat")) + if self.consumer is not None: + loops.append(asyncio.create_task(self.consumer.run(), name="consumer")) self._tasks = loops helpers: list[asyncio.Task[None]] = [] if self.notices is not None: @@ -206,15 +228,19 @@ async def ws_loop(self) -> None: await self._sleep(delay) async def _ingest_snapshot(self, payload: dict[str, Any], received_at: float) -> None: - """Fold one live-query snapshot into the inbox and log first sightings.""" - rows = [ - r for r in (payload.get("data") or {}).get(self.stream.entity) or [] if in_scope(r, self.scope, self.stream) - ] + """Fold one live-query snapshot into the inbox and log first sightings. + + A snapshot without the entity list raises (and the subscription reconnects) + instead of passing for an empty snapshot. + """ + rows = [r for r in entity_rows(payload, self.stream.entity) if in_scope(r, self.scope, self.stream)] clock = time.perf_counter() result = await asyncio.to_thread(self.inbox.observe, rows, SOURCE_WS, received_at) ingest_ms = (time.perf_counter() - clock) * 1000 self.ws_messages += 1 self.last_ws_message_at = received_at + if self.consumer is not None and (result.new or result.changed or result.restored): + self.consumer.wake() if self.notices is not None: await self.notices.new_events(SOURCE_WS, result.new_ids) self.notices.snapshot_changes(result.changed_ids) @@ -258,6 +284,8 @@ async def reconcile_loop(self) -> None: report = ReconcileReport(started_at=started, error=f"{type(exc).__name__}: {exc}") self.last_reconcile = report self.last_reconcile_at = self.now() + if self.consumer is not None and any(c.new or c.changed or c.restored for c in report.chains): + self.consumer.wake() ws_expected = ws_was_connected and self.config.transport == "both" self._log_reconcile(report, ws_expected) if self.notices is not None: diff --git a/utils/telegram.py b/utils/telegram.py index 8626ce84..751eac76 100644 --- a/utils/telegram.py +++ b/utils/telegram.py @@ -139,7 +139,7 @@ def send_telegram_message( origin_protocol: str | None = None, channel: str | None = None, metadata: dict[str, object] | None = None, -) -> None: +) -> bool: """ Send a message to a Telegram chat using a bot. @@ -148,6 +148,11 @@ def send_telegram_message( protocol: Protocol identifier used to select bot token and chat ID disable_notification: If True, sends the message silently + Returns: + True when Telegram accepted the message; False when it was skipped + (``LOG_LEVEL=DEBUG`` or missing credentials). Callers that record a + message as delivered must check this. + Raises: TelegramError: If the message fails to send """ @@ -175,7 +180,7 @@ def send_telegram_message( metadata=metadata, ) logger.debug("Skipping Telegram send (LOG_LEVEL=DEBUG)") - return + return False # Test/staging override: route every message to one chat for comparison runs. # Set TELEGRAM_TEST_CHAT_ID to a dummy group id; unset to restore prod routing. @@ -198,7 +203,7 @@ def send_telegram_message( metadata=metadata, ) logger.warning("TELEGRAM_TEST_CHAT_ID set but TELEGRAM_BOT_TOKEN_DEFAULT missing") - return + return False # Escape the label for Markdown sends — protocol names contain `_` # (e.g. yearn_timelock) and the brackets are link syntax in V1, either of # which would trip a 400 parse error and drop the comparison alert. The @@ -223,7 +228,7 @@ def send_telegram_message( _update_alert_delivery_safe(alert_id, status="failed", error=str(exc)) raise _update_alert_delivery_safe(alert_id, status="delivered", delivered_at=store.utc_now_iso()) - return + return True # Check if this protocol has a topic ID configured (forum-style group) topic_id = os.getenv(f"TELEGRAM_TOPIC_ID_{protocol.upper()}") @@ -252,7 +257,7 @@ def send_telegram_message( metadata=metadata, ) logger.warning("Missing Telegram credentials for %s", protocol) - return + return False alert_id = _record_alert_safe( message=message, @@ -280,10 +285,11 @@ def send_telegram_message( _update_alert_delivery_safe(alert_id, status="failed", error=str(retry_exc)) raise _update_alert_delivery_safe(alert_id, status="delivered", delivered_at=store.utc_now_iso()) - return + return True _update_alert_delivery_safe(alert_id, status="failed", error=str(exc)) raise _update_alert_delivery_safe(alert_id, status="delivered", delivered_at=store.utc_now_iso()) + return True def _record_alert_safe(