From d12dfe58af0d038370de33918b73da25accd1a26 Mon Sep 17 00:00:00 2001 From: spalen0 Date: Mon, 26 Oct 2026 17:16:22 +0100 Subject: [PATCH 1/2] feat(subscriber): named event streams with a Morpho governance stream Generalize the continuous timelock subscriber so one process follows several Envio entities (#447, part of #437): - Move the subscriber to utils/subscriber/. An EventStream names the entity, its address field, field selection, SQLite file and table prefix; the inbox, reconcile, worker, notices and report take it from the inbox instead of hardcoding TimelockEvent. - Timelock keeps timelock_subscriber.db and its timelock_sub_* tables. Schema v2 renames timelock_address to contract_address in place. - Add the morpho stream (MorphoGovernanceEvent from yearn-envio#71) for every configured V1/V2 vault, in morpho_subscriber.db. Observation only: no alerts, daily governance jobs stay the alerting path. - run_workers runs every stream in one process: one READY, a watchdog ping only while every stream is alive, one HTTP session per thread, and one chain_metadata read per cycle shared by all streams. - EVENT_SUBSCRIBER_STREAMS (default timelock) selects streams; run, reconcile-once, report and health cover each. Per-stream notice labels ([timelock-subscriber], [morpho-subscriber]). - Reorg overlap and block span for HyperEVM (999) and Robinhood (4663). Co-Authored-By: Claude Opus 5.5 --- .env.example | 3 +- deploy/install.sh | 4 +- deploy/runbook.md | 44 ++- .../monitoring-timelock-subscriber.service | 15 +- protocols/morpho/README.md | 8 + protocols/morpho/governance_stream.py | 74 ++++ protocols/timelock/README.md | 17 +- protocols/timelock/event_stream.py | 55 +++ protocols/timelock/event_subscriber.py | 191 ++++++--- tests/test_event_subscriber_streams.py | 374 ++++++++++++++++++ tests/test_timelock_subscriber_notices.py | 38 +- tests/test_timelock_subscriber_reconcile.py | 51 ++- tests/test_timelock_subscriber_worker.py | 43 +- tests/timelock_subscriber_fakes.py | 68 +++- utils/http_client.py | 5 +- .../timelock => utils}/subscriber/envio.py | 108 +++-- .../timelock => utils}/subscriber/inbox.py | 166 ++++---- .../timelock => utils}/subscriber/messages.py | 44 ++- .../timelock => utils}/subscriber/notices.py | 32 +- .../timelock => utils}/subscriber/notifier.py | 28 +- .../subscriber/reconcile.py | 56 ++- .../timelock => utils}/subscriber/report.py | 10 +- utils/subscriber/stream.py | 64 +++ .../timelock => utils}/subscriber/worker.py | 143 +++++-- 24 files changed, 1282 insertions(+), 359 deletions(-) create mode 100644 protocols/morpho/governance_stream.py create mode 100644 protocols/timelock/event_stream.py create mode 100644 tests/test_event_subscriber_streams.py rename {protocols/timelock => utils}/subscriber/envio.py (64%) rename {protocols/timelock => utils}/subscriber/inbox.py (76%) rename {protocols/timelock => utils}/subscriber/messages.py (71%) rename {protocols/timelock => utils}/subscriber/notices.py (90%) rename {protocols/timelock => utils}/subscriber/notifier.py (75%) rename {protocols/timelock => utils}/subscriber/reconcile.py (80%) rename {protocols/timelock => utils}/subscriber/report.py (96%) create mode 100644 utils/subscriber/stream.py rename {protocols/timelock => utils}/subscriber/worker.py (70%) diff --git a/.env.example b/.env.example index 7a9e04e8..d20fb74c 100644 --- a/.env.example +++ b/.env.example @@ -20,7 +20,8 @@ PROVIDER_URL_KATANA_1=https://rpc.katanarpc.com # Yearn large TVL env vars ENVIO_GRAPHQL_URL="https://indexer.hyperindex.xyz/3fec0a4/v1/graphql" -# Timelock event subscriber pilot (protocols/timelock/event_subscriber.py, local output only) +# Event subscriber pilot (protocols/timelock/event_subscriber.py, local output only) +# EVENT_SUBSCRIBER_STREAMS=timelock # comma-separated: timelock, morpho (needs MorphoGovernanceEvent in Envio) # 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/install.sh b/deploy/install.sh index 1691116c..635e53ae 100755 --- a/deploy/install.sh +++ b/deploy/install.sh @@ -243,8 +243,8 @@ cat <_subscriber.db and journald, +# and sends no alerts. TIMELOCK_SUBSCRIBER_NOTIFY=errors in the env file adds +# diagnostic notices to the internal errors chat. See +# protocols/timelock/README.md and deploy/runbook.md. # # Type=notify + WatchdogSec: the worker sends READY=1 once its loops start and # WATCHDOG=1 from its heartbeat only while HTTP reconciliation keeps completing, @@ -43,7 +44,7 @@ ExecStartPre=+/bin/bash -c '\ exit 1; \ fi' -# ENVIO_GRAPHQL_URL comes from here; TIMELOCK_SUBSCRIBER_* override the defaults. +# ENVIO_GRAPHQL_URL comes from here; EVENT_SUBSCRIBER_STREAMS and TIMELOCK_SUBSCRIBER_* override the defaults. EnvironmentFile=__ETC_DIR__/.env ExecStart=__REPO_DIR__/.venv/bin/python protocols/timelock/event_subscriber.py run diff --git a/protocols/morpho/README.md b/protocols/morpho/README.md index 65006ea9..e46be988 100644 --- a/protocols/morpho/README.md +++ b/protocols/morpho/README.md @@ -199,6 +199,14 @@ Both scripts share `MORPHO_FILENAME` (default `cache-id.txt`). New key types add | `v2_role` | `owner` / `curator` | lowercase address | Last-known instant-role address | | `v2_set` | `sentinels` / `allocators` / `adapters` | comma-joined lowercase addresses | Last-known role set | +### Continuous governance event stream (pilot) + +[`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). + +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. + ### Limitations `MorphoMarketV1AdapterV2` has its own internal timelock system (`setSkimRecipient`, `burnShares`, `increaseTimelock`, etc.). The Morpho GraphQL API does **not** expose adapter-internal pending operations, and replaying `Submit` events on each adapter would reintroduce the RPC cost we're explicitly avoiding. Phase-2 candidate. diff --git a/protocols/morpho/governance_stream.py b/protocols/morpho/governance_stream.py new file mode 100644 index 00000000..43f2e733 --- /dev/null +++ b/protocols/morpho/governance_stream.py @@ -0,0 +1,74 @@ +"""The ``MorphoGovernanceEvent`` stream followed by the continuous event subscriber. + +yearn-envio indexes governance events emitted by every vault in ``config.py`` +(V1 ``SubmitCap`` / ``SubmitMarketRemoval`` / ``SubmitTimelock``; V2 ``Submit`` / +``Accept`` / ``SetOwner`` / ``SetCurator`` / ``AddAdapter`` / ``RemoveAdapter``) as +one entity with ABI-named arguments in ``argsJson``. This stream observes them +with the same WebSocket + HTTP reconciliation as the timelock pilot (#447). It +sends no alerts: lifecycle rules and delivery are #448, and the daily +``governance.py`` / ``governance_v2.py`` jobs stay the alerting path. +""" + +import json +from typing import Any + +from protocols.morpho._shared import PROTOCOL +from protocols.morpho.config import VAULTS_V1_BY_CHAIN, VAULTS_V2_BY_CHAIN, iter_vaults +from utils.subscriber.stream import EventStream, StreamLabel, StreamScope + +MORPHO_GOVERNANCE_FIELDS = """ + id + vaultAddress + vaultVersion + eventName + chainId + blockNumber + blockHash + blockTimestamp + logIndex + transactionHash + transactionIndex + transactionFrom + argsJson +""" + +# Longest ``argsJson`` shown in a notice; V2 ``Submit`` carries full calldata. +_MAX_ARGS_CHARS = 600 + + +def describe_governance_event(payload: dict[str, Any]) -> list[str]: + """Return notice lines for one event: vault generation, sender, and its ABI-named arguments.""" + lines = [f"Vault V{payload.get('vaultVersion')}, sent by {payload.get('transactionFrom') or 'n/a'}"] + raw = payload.get("argsJson") + if raw: + try: + args = json.dumps(json.loads(raw), separators=(",", ":"), sort_keys=True) + except TypeError, ValueError: + args = str(raw) + lines.append(f"Args: {args[:_MAX_ARGS_CHARS]}" + ("…" if len(args) > _MAX_ARGS_CHARS else "")) + return lines + + +MORPHO_GOVERNANCE_STREAM = EventStream( + name="morpho", + noun="Morpho governance", + entity="MorphoGovernanceEvent", + address_field="vaultAddress", + address_label="Vault", + fields=MORPHO_GOVERNANCE_FIELDS, + db_filename="morpho_subscriber.db", + table_prefix="morpho_sub", + describe=describe_governance_event, +) + + +def morpho_governance_scope() -> StreamScope: + """Return every configured V1 and V2 vault, keyed by chain, with its notice label.""" + addresses: dict[int, set[str]] = {} + labels: dict[tuple[str, int], StreamLabel] = {} + for version, vaults_by_chain in (("V1", VAULTS_V1_BY_CHAIN), ("V2", VAULTS_V2_BY_CHAIN)): + for chain, vault in iter_vaults(vaults_by_chain): + address = vault.address.lower() + addresses.setdefault(chain.chain_id, set()).add(address) + labels[(address, chain.chain_id)] = StreamLabel(f"{vault.name} {version}", PROTOCOL) + return StreamScope(addresses=addresses, labels=labels) diff --git a/protocols/timelock/README.md b/protocols/timelock/README.md index 0b298620..795c71d0 100644 --- a/protocols/timelock/README.md +++ b/protocols/timelock/README.md @@ -15,12 +15,21 @@ The script runs hourly via the [monitoring runner](../../automation/jobs.yaml). `event_subscriber.py` is a continuously running worker that observes the same `TimelockEvent` rows as `timelock_alerts.py` with lower latency ([#438](https://github.com/yearn/monitoring/issues/438), the first step of [#437](https://github.com/yearn/monitoring/issues/437)). It is an observation pilot: **it sends no alerts and changes nothing about current delivery**. The hourly task remains the only sender to protocol groups. Optionally it posts diagnostic notices to the internal errors chat (see [Pilot notices](#pilot-notices)). It runs as the `monitoring-timelock-subscriber` systemd unit; operations are in the [runbook](../../deploy/runbook.md#timelock-event-subscriber-pilot). -It runs two transports side by side and records which one saw each event first: +The worker follows **named event streams** ([#447](https://github.com/yearn/monitoring/issues/447)). The shared code lives in [`utils/subscriber/`](../../utils/subscriber/); each stream is an `EventStream` that names its Envio entity, the field holding the monitored contract, its field selection, and its own SQLite file and table prefix: + +| Stream | Entity | Defined in | Inbox | +|---|---|---|---| +| `timelock` (default) | `TimelockEvent` | [`event_stream.py`](./event_stream.py) | `timelock_subscriber.db`, tables `timelock_sub_*` | +| `morpho` | `MorphoGovernanceEvent` | [`protocols/morpho/governance_stream.py`](../morpho/governance_stream.py) | `morpho_subscriber.db`, tables `morpho_sub_*` | + +`EVENT_SUBSCRIBER_STREAMS` (or `--streams`, before the subcommand) picks the streams, e.g. `timelock,morpho`; the default is `timelock` alone, so an existing deployment keeps its behavior. All selected streams run in one process (`run_workers`): one HTTP poster with a persistent session per thread, one `chain_metadata` read per cycle shared by every stream (`SharedChainHeads`), one WebSocket connection and one reconcile loop per stream, and a systemd watchdog ping only while every stream is alive. A stream's failure (for example an entity the Envio deployment does not serve yet) shows in its own log, checkpoints, and health, and never blocks another stream. The `run`, `reconcile-once`, `report`, and `health` subcommands cover every selected stream. + +Each stream runs two transports side by side and records which one saw each event first: - **WebSocket**: a `graphql-transport-ws` live query (`utils/graphql_ws.py`) over the newest 50 monitored events, filtered by monitored chains and addresses. Hasura resends the whole snapshot on every change, so most rows are repeats; the inbox deduplicates them. On disconnect the worker reconnects with capped exponential backoff (1s to 60s). - **HTTP reconciliation**, every 60s and also while the WebSocket is connected: for each chain, read `chain_metadata.latest_processed_block`, then fetch `(checkpoint - overlap, head]` in chunks of at most `MAX_BLOCK_SPAN` blocks, paginated in pages of 200 ordered by `(blockNumber, logIndex, id)`. This is the completeness guarantee. -State lives in `$CACHE_DIR/timelock_subscriber.db` (SQLite, WAL), separate from the hourly monitor's state. The schema is versioned (`PRAGMA user_version`); a database from an earlier revision is migrated in place on open, keeping observations, changes, and checkpoints. A row stored before the event selection grew (e.g. before `blockHash`) is upgraded silently on its next observation, not recorded as a change. +State lives in the stream's file under `$CACHE_DIR` (SQLite, WAL), separate from the scheduled monitors' state. The tables below are named for the timelock stream; other streams use their own prefix. The schema is versioned (`PRAGMA user_version`); a database from an earlier revision is migrated in place on open, keeping observations, changes, and checkpoints (v1 → v2 renamed `timelock_address` to `contract_address`). A row stored before the event selection grew (e.g. before `blockHash`) is upgraded silently on its next observation, not recorded as a change. - **Per-chain checkpoints** (`timelock_sub_checkpoints`): the highest block fully fetched over HTTP, never moved backwards. They are independent per chain and do not use `TIMELOCK_LAST_TS`, so a failing or lagging chain never holds back the others. A chain's first run starts at the indexed head unless `TIMELOCK_SUBSCRIBER_START_LOOKBACK_SECONDS` is set. - **Event provenance** (`timelock_sub_events`): one row per Envio event id (`__`) with its block hash, a content fingerprint, the first transport and time it was seen, first-seen times and observation counts per transport. @@ -36,7 +45,7 @@ Edge cases: Supervision: a heartbeat every 30s stores worker status, flags any chain whose indexed head has not advanced for 15 minutes, and pings the systemd watchdog only while reconciliation keeps completing. `event_subscriber.py health` exits 1 on a stale heartbeat, failed reconcile, WebSocket outage, or stalled chain. -`event_subscriber.py report --hours N` prints the coverage/latency report used to choose the production transport: +`event_subscriber.py report --hours N` prints each selected stream's coverage/latency report, used to choose the production transport: - WebSocket coverage of HTTP-confirmed events, plus WebSocket-only and HTTP-only counts. - Delay from block timestamp to first sighting per transport (p50/p95/max), and how far the WebSocket leads HTTP. Envio exposes no per-row index time, so the faster transport's delay is an upper bound on indexing delay. @@ -46,7 +55,7 @@ Supervision: a heartbeat every 30s stores worker status, flags any chain whose i ### Pilot notices -With `TIMELOCK_SUBSCRIBER_NOTIFY=errors` (or `run --notify errors`), the worker posts plain-text diagnostics to the internal errors chat (`TELEGRAM_TOPIC_ID_ERRORS` or `TELEGRAM_CHAT_ID_ERRORS`), labelled `[timelock-subscriber]` and stored in alert history under `yearn-internal`. They are for comparing the pilot with the hourly alerts, not alerts themselves. Notices stay off unless an errors chat is configured: `send_error_message` would otherwise fall back to a protocol group. +With `TIMELOCK_SUBSCRIBER_NOTIFY=errors` (or `run --notify errors`), the worker posts plain-text diagnostics to the internal errors chat (`TELEGRAM_TOPIC_ID_ERRORS` or `TELEGRAM_CHAT_ID_ERRORS`), labelled `[timelock-subscriber]` and stored in alert history under `yearn-internal`. Each stream posts under its own label (`[timelock-subscriber]`, `[morpho-subscriber]`), and its texts name the stream ("New Morpho governance event"). They are for comparing the pilot with the scheduled alerts, not alerts themselves. Notices stay off unless an errors chat is configured: `send_error_message` would otherwise fall back to a protocol group. | Notice | When | |---|---| diff --git a/protocols/timelock/event_stream.py b/protocols/timelock/event_stream.py new file mode 100644 index 00000000..33b74e70 --- /dev/null +++ b/protocols/timelock/event_stream.py @@ -0,0 +1,55 @@ +"""The ``TimelockEvent`` stream followed by the continuous event subscriber.""" + +from protocols.timelock.timelock_alerts import TIMELOCK_LIST +from utils.subscriber.reconcile import build_scope +from utils.subscriber.stream import EventStream, StreamLabel, StreamScope + +# The hourly monitor's selection plus logIndex and blockHash. Both transports use +# it, so a row fingerprints identically whichever one fetched it, and a reorg that +# replaces the block changes the fingerprint through blockHash. +TIMELOCK_FIELDS = """ + id + timelockAddress + timelockType + eventName + chainId + blockNumber + blockHash + blockTimestamp + logIndex + transactionHash + operationId + index + target + value + data + predecessor + delay + signature + creator + metadata + votesFor + votesAgainst +""" + +# Database file and table names predate named streams; keeping them lets an +# existing pilot inbox continue with its checkpoints and history. +TIMELOCK_STREAM = EventStream( + name="timelock", + noun="timelock", + entity="TimelockEvent", + address_field="timelockAddress", + address_label="Timelock", + fields=TIMELOCK_FIELDS, + db_filename="timelock_subscriber.db", + table_prefix="timelock_sub", + operation_field="operationId", +) + + +def timelock_scope() -> StreamScope: + """Return every monitored timelock, keyed by chain, with its notice label.""" + return StreamScope( + addresses=build_scope(TIMELOCK_LIST), + labels={(t.address.lower(), t.chain_id): StreamLabel(t.label, t.protocol) for t in TIMELOCK_LIST}, + ) diff --git a/protocols/timelock/event_subscriber.py b/protocols/timelock/event_subscriber.py index 6728231d..c5524bb2 100644 --- a/protocols/timelock/event_subscriber.py +++ b/protocols/timelock/event_subscriber.py @@ -1,17 +1,22 @@ #!/usr/bin/env python3 -"""Continuous timelock event subscriber (observation pilot). +"""Continuous event subscriber (observation pilot) for named Envio event streams. -Runs alongside the hourly ``timelock_alerts.py`` and never replaces it: it -observes the same ``TimelockEvent`` rows over an Envio WebSocket subscription and -bounded HTTP reconciliation, records provenance and timing in a local SQLite -inbox, and sends no alerts. With ``--notify errors`` it posts diagnostic notices -to the internal errors chat. See ``protocols/timelock/README.md``. +Runs alongside the scheduled monitors and never replaces them: for each selected +stream (``timelock``: ``TimelockEvent``; ``morpho``: ``MorphoGovernanceEvent``) it +observes the entity over an Envio WebSocket subscription and bounded HTTP +reconciliation, records provenance and timing in the stream's own SQLite inbox, +and sends no alerts. With ``--notify errors`` it posts diagnostic notices to the +internal errors chat. See ``protocols/timelock/README.md``. + +The script lives here because the timelock pilot came first and its systemd unit +runs this path; every stream shares the process, the HTTP session, and the +chain-head reads. Usage: - python protocols/timelock/event_subscriber.py run [--transport both|ws|http] - python protocols/timelock/event_subscriber.py reconcile-once - python protocols/timelock/event_subscriber.py report [--hours 24] [--json] - python protocols/timelock/event_subscriber.py health + python protocols/timelock/event_subscriber.py [--streams timelock,morpho] run [--transport both|ws|http] + python protocols/timelock/event_subscriber.py [--streams ...] reconcile-once + python protocols/timelock/event_subscriber.py [--streams ...] report [--hours 24] [--json] + python protocols/timelock/event_subscriber.py [--streams ...] health """ import argparse @@ -20,32 +25,64 @@ import os import sys import time +from collections.abc import Callable, Mapping from dataclasses import asdict +from typing import Any from dotenv import load_dotenv -from protocols.timelock.subscriber.envio import http_poster -from protocols.timelock.subscriber.inbox import EventInbox -from protocols.timelock.subscriber.notices import PilotNotices -from protocols.timelock.subscriber.notifier import NOTIFY_MODES, build_notifier -from protocols.timelock.subscriber.reconcile import ReconcileConfig, build_scope, reconcile_all -from protocols.timelock.subscriber.report import build_report, format_report, report_to_json -from protocols.timelock.subscriber.worker import TRANSPORTS, Worker, WorkerConfig -from protocols.timelock.timelock_alerts import TIMELOCK_LIST, TIMELOCKS +from protocols.morpho.governance_stream import MORPHO_GOVERNANCE_STREAM, morpho_governance_scope +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 +from utils.subscriber.envio import SharedChainHeads, http_poster +from utils.subscriber.inbox import EventInbox +from utils.subscriber.notices import PilotNotices +from utils.subscriber.notifier import NOTIFY_MODES, build_notifier +from utils.subscriber.reconcile import ReconcileConfig, reconcile_all +from utils.subscriber.report import build_report, format_report, report_to_json +from utils.subscriber.stream import EventStream, StreamScope +from utils.subscriber.worker import TRANSPORTS, Worker, WorkerConfig, run_workers load_dotenv() _logger = get_logger("timelock.event_subscriber") +# Streams this process can follow. Morpho needs an Envio deployment that serves +# MorphoGovernanceEvent (yearn-envio#71); until then leave it out of --streams. +STREAMS: dict[str, tuple[EventStream, Callable[[], StreamScope]]] = { + TIMELOCK_STREAM.name: (TIMELOCK_STREAM, timelock_scope), + MORPHO_GOVERNANCE_STREAM.name: (MORPHO_GOVERNANCE_STREAM, morpho_governance_scope), +} +DEFAULT_STREAMS = TIMELOCK_STREAM.name + + +def parse_streams(value: str) -> list[str]: + """Parse a comma-separated stream list, keeping order and dropping duplicates. + + Raises: + argparse.ArgumentTypeError: A name is not in ``STREAMS`` or the list is empty. + """ + names = list(dict.fromkeys(name.strip() for name in value.split(",") if name.strip())) + unknown = [name for name in names if name not in STREAMS] + if unknown or not names: + raise argparse.ArgumentTypeError(f"streams must be a non-empty subset of {sorted(STREAMS)}, got {value!r}") + return names + + +def _selected(args: argparse.Namespace) -> list[tuple[EventStream, StreamScope]]: + """Return each selected stream with its monitored scope.""" + return [(STREAMS[name][0], STREAMS[name][1]()) for name in args.streams] + def _env_float(name: str, default: float) -> float: + """Read a float environment variable, falling back to ``default`` when unset or empty.""" value = os.getenv(name) return float(value) if value else default def _graphql_url() -> str: + """Return ``ENVIO_GRAPHQL_URL`` or exit.""" url = os.getenv("ENVIO_GRAPHQL_URL") if not url: raise SystemExit("ENVIO_GRAPHQL_URL is not set") @@ -53,6 +90,7 @@ def _graphql_url() -> str: def _reconcile_config(args: argparse.Namespace) -> ReconcileConfig: + """Build the reconcile bounds from the CLI arguments.""" return ReconcileConfig( page_size=args.page_size, max_chunks_per_cycle=args.max_chunks, @@ -61,7 +99,7 @@ def _reconcile_config(args: argparse.Namespace) -> ReconcileConfig: def cmd_run(args: argparse.Namespace) -> None: - """Run the worker until SIGTERM.""" + """Run one worker per selected stream in this process until SIGTERM.""" url = _graphql_url() ws_url = os.getenv("ENVIO_GRAPHQL_WS_URL") or http_to_ws_url(url) config = WorkerConfig( @@ -72,65 +110,96 @@ def cmd_run(args: argparse.Namespace) -> None: head_stall_seconds=args.head_stall_seconds, reconcile=_reconcile_config(args), ) - inbox = EventInbox() - notifier = build_notifier(args.notify) - notices = PilotNotices(notifier, inbox, TIMELOCKS) if notifier is not None else None - worker = Worker(http_poster(url), ws_url, inbox, build_scope(TIMELOCK_LIST), config, notices=notices) - _logger.info( - "starting timelock subscriber: transport=%s reconcile every %.0fs notices=%s", - config.transport, - config.reconcile_interval, - "errors chat" if notices else "off", - ) - asyncio.run(worker.run()) + post = http_poster(url) + # One chain_metadata read per cycle for every stream. + heads = SharedChainHeads(post) + workers = [] + for stream, scope in _selected(args): + 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)) + _logger.info( + "starting %s stream: %s chains, transport=%s reconcile every %.0fs notices=%s", + stream.name, + len(scope.addresses), + config.transport, + config.reconcile_interval, + "errors chat" if notices else "off", + ) + asyncio.run(run_workers(workers, watchdog_interval=config.heartbeat_interval)) def cmd_reconcile_once(args: argparse.Namespace) -> None: - """Run one reconcile cycle and print its per-chain result.""" - report = reconcile_all( - http_poster(_graphql_url()), EventInbox(), build_scope(TIMELOCK_LIST), _reconcile_config(args) - ) - summary = { - "ok": report.ok, - "error": report.error, - "duration_ms": round(report.duration_ms, 1), - "chains": [asdict(c) for c in report.chains], - } - print(json.dumps(summary, indent=2, default=str)) - if not report.ok: + """Run one reconcile cycle per selected stream and print each per-chain result.""" + post = http_poster(_graphql_url()) + heads = SharedChainHeads(post) + summaries = {} + ok = True + for stream, scope in _selected(args): + report = reconcile_all(post, EventInbox(stream), scope.addresses, _reconcile_config(args), heads=heads) + ok = ok and report.ok + summaries[stream.name] = { + "ok": report.ok, + "error": report.error, + "duration_ms": round(report.duration_ms, 1), + "chains": [asdict(c) for c in report.chains], + } + print(json.dumps(summaries, indent=2, default=str)) + if not ok: sys.exit(1) def cmd_report(args: argparse.Namespace) -> None: - """Print the coverage/latency report.""" - report = build_report(EventInbox(), args.hours) - print(json.dumps(report_to_json(report), indent=2) if args.json else format_report(report)) + """Print the coverage/latency report of every selected stream.""" + 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())) + + +def heartbeat_problems(status: Mapping[str, tuple[Any, float]], now: float) -> list[str]: + """Return what is wrong with one stream's stored heartbeat, if anything.""" + heartbeat, updated_at = status.get("heartbeat", (None, 0.0)) + if not isinstance(heartbeat, dict): + return ["no heartbeat recorded"] + problems: list[str] = [] + age = now - updated_at + if age > 3 * float(heartbeat.get("heartbeat_interval") or 30): + problems.append(f"heartbeat is {age:.0f}s old") + if heartbeat.get("last_reconcile_ok") is False: + problems.append("last reconcile failed") + if heartbeat.get("transport") in ("both", "ws") and not heartbeat.get("ws_connected"): + problems.append("WebSocket disconnected") + problems += [f"chain {c['chain_id']} head stalled" for c in heartbeat.get("chains", []) if c.get("stalled")] + return problems def cmd_health(args: argparse.Namespace) -> None: - """Exit non-zero when the worker's heartbeat is stale, reconciliation failed, or a chain stalled.""" - status = EventInbox().status() + """Exit non-zero when any selected stream's heartbeat is stale, reconciliation failed, or a chain stalled.""" + now = time.time() + streams = {} problems: list[str] = [] - heartbeat, updated_at = status.get("heartbeat", (None, 0.0)) - if heartbeat is None: - problems.append("no heartbeat recorded") - else: - age = time.time() - updated_at - if age > 3 * float(heartbeat.get("heartbeat_interval") or 30): - problems.append(f"heartbeat is {age:.0f}s old") - if heartbeat.get("last_reconcile_ok") is False: - problems.append("last reconcile failed") - if heartbeat.get("transport") in ("both", "ws") and not heartbeat.get("ws_connected"): - problems.append("WebSocket disconnected") - problems += [f"chain {c['chain_id']} head stalled" for c in heartbeat.get("chains", []) if c.get("stalled")] - print(json.dumps({"ok": not problems, "problems": problems, "heartbeat": heartbeat}, indent=2)) + for stream, _ in _selected(args): + status = EventInbox(stream).status() + stream_problems = heartbeat_problems(status, 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)) if problems: sys.exit(1) def main() -> None: """Parse arguments and dispatch the subcommand.""" - parser = argparse.ArgumentParser(description="Continuous timelock event subscriber (observation pilot).") + parser = argparse.ArgumentParser(description="Continuous event subscriber (observation pilot).") + parser.add_argument( + "--streams", + type=parse_streams, + default=os.getenv("EVENT_SUBSCRIBER_STREAMS", DEFAULT_STREAMS), + help=f"Comma-separated streams to follow, from {sorted(STREAMS)} (default: {DEFAULT_STREAMS}).", + ) sub = parser.add_subparsers(dest="command", required=True) def add_reconcile_args(p: argparse.ArgumentParser) -> None: diff --git a/tests/test_event_subscriber_streams.py b/tests/test_event_subscriber_streams.py new file mode 100644 index 00000000..cb2cfa85 --- /dev/null +++ b/tests/test_event_subscriber_streams.py @@ -0,0 +1,374 @@ +"""Tests for named event streams: Morpho governance alongside timelocks in one subscriber process.""" + +import argparse +import asyncio +import json +import sqlite3 +import threading +import time +from pathlib import Path +from unittest.mock import patch + +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 tests.timelock_subscriber_fakes import ( + MORPHO_VAULT, + TIMELOCK, + FakeEnvio, + FakeWSServer, + make_event, + 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.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.worker import Worker, WorkerConfig, run_workers + +TIMELOCK_SCOPE = {1: {TIMELOCK}} +MORPHO_SCOPE = {1: {MORPHO_VAULT}} +CONFIG = ReconcileConfig(page_size=20, overlap_blocks={1: 5}, max_block_span={1: 50}) + + +class RecordingNotifier(Notifier): + """Notifier that keeps posted notices instead of sending them.""" + + def __init__(self) -> None: + super().__init__(send=lambda text: None) + self.posted: list[str] = [] + + def post(self, text: str) -> None: + self.posted.append(text) + + +def _inboxes(tmp_path: Path) -> tuple[EventInbox, EventInbox]: + return EventInbox(TIMELOCK_STREAM, str(tmp_path / "timelock.db")), EventInbox( + MORPHO_GOVERNANCE_STREAM, str(tmp_path / "morpho.db") + ) + + +def _ids(inbox: EventInbox) -> set[str]: + return {row["event_id"] for row in inbox.events_since(0)} + + +# ── 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"}'] + + long = describe_governance_event(make_morpho_event(1, 10, argsJson=json.dumps({"data": "0x" + "ab" * 1000}))) + assert long[1].endswith("…") and len(long[1]) < 700 + assert describe_governance_event(make_morpho_event(1, 10, argsJson="not json"))[1] == "Args: not json" + + +# ── isolated inboxes and shared chain heads ──────────────────────────── + + +def test_streams_reconcile_into_separate_inboxes_with_one_metadata_read(tmp_path: Path) -> None: + envio = FakeEnvio() + envio.set_head(1, 1) + timelock_inbox, morpho_inbox = _inboxes(tmp_path) + heads = SharedChainHeads(envio.post) + assert reconcile_all(envio.post, timelock_inbox, TIMELOCK_SCOPE, CONFIG, heads=heads).ok + assert reconcile_all(envio.post, morpho_inbox, MORPHO_SCOPE, CONFIG, heads=heads).ok + + envio.add(make_event(1, 10), make_morpho_event(1, 11), make_morpho_event(1, 12, address="0x" + "99" * 20)) + shared = SharedChainHeads(envio.post) + reads = envio.metadata_reads + timelock = reconcile_all(envio.post, timelock_inbox, TIMELOCK_SCOPE, CONFIG, heads=shared) + morpho = reconcile_all(envio.post, morpho_inbox, MORPHO_SCOPE, CONFIG, heads=shared) + + assert envio.metadata_reads == reads + 1 # two streams, one chain_metadata read + assert (timelock.chains[0].new, morpho.chains[0].new) == (1, 1) + assert _ids(timelock_inbox) == {"1_10_0"} + assert _ids(morpho_inbox) == {"1_11_0"} # the unmonitored vault stays out + (row,) = morpho_inbox.events_since(0) + assert row["contract_address"] == MORPHO_VAULT and row["operation_id"] is None + with sqlite3.connect(morpho_inbox.path) as conn: + tables = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")} + assert "morpho_sub_events" in tables and not any(t.startswith("timelock_sub") for t in tables) + + +def test_streams_keep_independent_checkpoints(tmp_path: Path) -> None: + envio = FakeEnvio() + envio.set_head(1, 1) + timelock_inbox, morpho_inbox = _inboxes(tmp_path) + reconcile_all(envio.post, timelock_inbox, TIMELOCK_SCOPE, CONFIG) + reconcile_all(envio.post, morpho_inbox, MORPHO_SCOPE, CONFIG) + envio.add(make_morpho_event(1, 30)) + envio.fail_chain[1] = 1 # the next range query (timelock's) fails + + assert not reconcile_all(envio.post, timelock_inbox, TIMELOCK_SCOPE, CONFIG).ok + assert reconcile_all(envio.post, morpho_inbox, MORPHO_SCOPE, CONFIG).ok + + timelock_checkpoint, morpho_checkpoint = timelock_inbox.get_checkpoint(1), morpho_inbox.get_checkpoint(1) + assert timelock_checkpoint is not None and morpho_checkpoint is not None + assert (timelock_checkpoint.block_number, morpho_checkpoint.block_number) == (1, 30) + + +def test_shared_chain_heads_refetch_when_stale_and_retry_after_errors() -> None: + envio = FakeEnvio() + envio.set_head(1, 5) + clock = [0.0] + heads = SharedChainHeads(envio.post, max_age=10.0, clock=lambda: clock[0]) + + assert heads()[1].latest_processed_block == 5 + envio.set_head(1, 6) + clock[0] = 9.0 + assert heads()[1].latest_processed_block == 5 # still fresh + clock[0] = 20.0 + envio.fail_metadata = 1 + with pytest.raises(EnvioQueryError): + heads() + assert heads()[1].latest_processed_block == 6 + assert envio.metadata_reads == 3 + + +def test_http_poster_keeps_one_session_per_thread() -> None: + sessions: list[object] = [] + + class Response: + def json(self) -> dict: + return {"data": {}} + + def fake_request(*_args: object, session: object, **_kwargs: object) -> Response: + sessions.append(session) + return Response() + + with patch.object(envio_module, "request_with_retry", side_effect=fake_request): + post = http_poster("https://envio.example/v1/graphql") + post("{ a }", {}) + post("{ b }", {}) + thread = threading.Thread(target=post, args=("{ c }", {})) + thread.start() + thread.join() + + assert sessions[0] is sessions[1] + assert sessions[2] is not sessions[0] + + +# ── notices and reports ──────────────────────────────────────────────── + + +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) + 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 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 ─────────────────────────────────────── + + +def _worker_config() -> WorkerConfig: + return WorkerConfig( + reconcile_interval=0.2, + heartbeat_interval=0.05, + backoff_initial=0.05, + backoff_max=0.2, + reconcile=CONFIG, + ) + + +async def _until(predicate, timeout: float = 5.0) -> None: # noqa: ANN001 + deadline = time.monotonic() + timeout + while not predicate(): + if time.monotonic() > deadline: + raise AssertionError("condition not met in time") + await asyncio.sleep(0.02) + + +def test_run_workers_follows_both_streams_over_websocket(tmp_path: Path) -> None: + notified: list[tuple[str, ...]] = [] + + async def scenario() -> None: + envio = FakeEnvio() + envio.set_head(1, 1) + server = FakeWSServer(envio) + await server.start() + timelock_inbox, morpho_inbox = _inboxes(tmp_path) + heads = SharedChainHeads(envio.post) + workers = [ + Worker(envio.post, server.url, timelock_inbox, TIMELOCK_SCOPE, _worker_config(), heads=heads), + Worker(envio.post, server.url, morpho_inbox, MORPHO_SCOPE, _worker_config(), heads=heads), + ] + task = asyncio.create_task(run_workers(workers, watchdog_interval=0.05)) + try: + await _until(lambda: all(w.ws_connected and w.last_reconcile_at is not None for w in workers)) + now = int(time.time()) + envio.add(make_event(1, 10, ts=now), make_morpho_event(1, 11, ts=now)) + await _until(lambda: _ids(timelock_inbox) == {"1_10_0"} and _ids(morpho_inbox) == {"1_11_0"}) + await asyncio.sleep(0.1) # let the supervisor ping the watchdog + finally: + workers[0].stop() # stopping one stream stops the process + await asyncio.wait_for(task, 5) + await server.stop() + + assert server.subscriptions >= 2 # one connection per stream + assert {row["first_source"] for row in timelock_inbox.events_since(0)} == {SOURCE_WS} + assert {row["first_source"] for row in morpho_inbox.events_since(0)} == {SOURCE_WS} + for inbox, name in ((timelock_inbox, "timelock"), (morpho_inbox, "morpho")): + heartbeat = inbox.status()["heartbeat"][0] + assert heartbeat["stream"] == name + + with patch("utils.subscriber.worker.notify", side_effect=lambda *a: notified.append(a)): + asyncio.run(scenario()) + + kinds = [message[0] for message in notified] + assert kinds.count("READY=1") == 1 and kinds.count("STOPPING=1") == 1 + assert "WATCHDOG=1" in kinds + watchdog_status = next(m[1] for m in notified if m[0] == "WATCHDOG=1") + assert "timelock:ws=" in watchdog_status and "morpho:ws=" in watchdog_status + + +def test_run_workers_withholds_the_watchdog_while_any_stream_is_dead(tmp_path: Path) -> None: + notified: list[tuple[str, ...]] = [] + + async def scenario() -> None: + envio = FakeEnvio() + envio.set_head(1, 1) + timelock_inbox, morpho_inbox = _inboxes(tmp_path) + config = WorkerConfig(transport="http", reconcile_interval=0.1, heartbeat_interval=0.05, reconcile=CONFIG) + workers = [ + Worker(envio.post, None, timelock_inbox, TIMELOCK_SCOPE, config), + Worker(envio.post, None, morpho_inbox, MORPHO_SCOPE, config), + ] + task = asyncio.create_task(run_workers(workers, watchdog_interval=0.02)) + try: + await _until(lambda: all(w.last_reconcile_at is not None for w in workers)) + with patch.object(workers[1], "is_alive", return_value=False): + notified.clear() + await asyncio.sleep(0.15) + assert all(m[0] != "WATCHDOG=1" for m in notified) + finally: + workers[0].stop() + await asyncio.wait_for(task, 5) + + with patch("utils.subscriber.worker.notify", side_effect=lambda *a: notified.append(a)): + asyncio.run(scenario()) + + +def test_run_workers_surfaces_a_crashed_stream(tmp_path: Path) -> None: + async def scenario() -> None: + envio = FakeEnvio() + timelock_inbox, morpho_inbox = _inboxes(tmp_path) + config = WorkerConfig(transport="http", reconcile_interval=60, heartbeat_interval=0.05, reconcile=CONFIG) + healthy = Worker(envio.post, None, timelock_inbox, TIMELOCK_SCOPE, config) + broken = Worker(envio.post, None, morpho_inbox, MORPHO_SCOPE, config) + + async def crash() -> None: + raise RuntimeError("boom") + + with patch.object(broken, "run", crash): + await asyncio.wait_for(run_workers([healthy, broken]), 5) + + with patch("utils.subscriber.worker.notify"), pytest.raises(RuntimeError, match="boom"): + 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 ──────────────────────────────────────────────────────────────── + + +def test_parse_streams_keeps_order_and_rejects_unknown_names() -> None: + assert parse_streams("timelock") == ["timelock"] + assert parse_streams(" morpho, timelock ,morpho") == ["morpho", "timelock"] + for bad in ("", "timelock,aave", " , "): + with pytest.raises(argparse.ArgumentTypeError): + parse_streams(bad) + + +def test_heartbeat_problems() -> None: + now = 1_000.0 + healthy = { + "heartbeat": ( + {"heartbeat_interval": 30, "last_reconcile_ok": True, "transport": "both", "ws_connected": True}, + now - 10, + ) + } + assert heartbeat_problems(healthy, now) == [] + assert heartbeat_problems({}, now) == ["no heartbeat recorded"] + stale = { + "heartbeat": ( + { + "heartbeat_interval": 30, + "last_reconcile_ok": False, + "transport": "both", + "ws_connected": False, + "chains": [{"chain_id": 999, "stalled": True}], + }, + now - 200, + ) + } + assert heartbeat_problems(stale, now) == [ + "heartbeat is 200s old", + "last reconcile failed", + "WebSocket disconnected", + "chain 999 head stalled", + ] diff --git a/tests/test_timelock_subscriber_notices.py b/tests/test_timelock_subscriber_notices.py index 8db7401f..b7bb0f84 100644 --- a/tests/test_timelock_subscriber_notices.py +++ b/tests/test_timelock_subscriber_notices.py @@ -7,19 +7,19 @@ import pytest -from protocols.timelock.subscriber.inbox import SOURCE_HTTP, SOURCE_WS, EventInbox -from protocols.timelock.subscriber.notices import NoticeConfig, PilotNotices -from protocols.timelock.subscriber.notifier import ( +from protocols.timelock.event_stream import TIMELOCK_STREAM +from protocols.timelock.timelock_alerts import TimelockConfig +from tests.timelock_subscriber_fakes import TIMELOCK, FakeEnvio, FakeWSServer, make_event +from utils.subscriber.inbox import SOURCE_HTTP, SOURCE_WS, EventInbox +from utils.subscriber.notices import NoticeConfig, PilotNotices +from utils.subscriber.notifier import ( ALERT_PROTOCOL, - NOTIFY_LABEL, Notifier, build_notifier, send_to_errors_chat, ) -from protocols.timelock.subscriber.reconcile import ChainReconcileResult, ReconcileConfig, ReconcileReport -from protocols.timelock.subscriber.worker import Worker, WorkerConfig -from protocols.timelock.timelock_alerts import TimelockConfig -from tests.timelock_subscriber_fakes import TIMELOCK, FakeEnvio, FakeWSServer, make_event +from utils.subscriber.reconcile import ChainReconcileResult, ReconcileConfig, ReconcileReport +from utils.subscriber.worker import Worker, WorkerConfig TIMELOCKS = {(TIMELOCK, 1): TimelockConfig(TIMELOCK, 1, "YEARN_TIMELOCK", "Yearn TimelockController")} SCOPE = {1: {TIMELOCK}, 8453: {TIMELOCK}} @@ -38,7 +38,7 @@ def post(self, text: str) -> None: @pytest.fixture def inbox(tmp_path: Path) -> EventInbox: - return EventInbox(str(tmp_path / "sub.db")) + return EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")) @pytest.fixture @@ -101,18 +101,20 @@ async def scenario() -> tuple[list[str], int]: def test_build_notifier_requires_errors_chat(monkeypatch: pytest.MonkeyPatch) -> None: - assert build_notifier("none") is None - assert build_notifier("errors") is None # conftest strips TELEGRAM_*: no errors chat + assert build_notifier("none", "timelock") is None + assert build_notifier("errors", "timelock") is None # conftest strips TELEGRAM_*: no errors chat monkeypatch.setenv("TELEGRAM_CHAT_ID_ERRORS", "-100123") - assert isinstance(build_notifier("errors"), Notifier) + assert isinstance(build_notifier("errors", "timelock"), Notifier) with pytest.raises(ValueError, match="notify mode"): - build_notifier("public") + build_notifier("public", "timelock") def test_send_to_errors_chat_is_labelled_and_internal() -> None: - with patch("protocols.timelock.subscriber.notifier.send_error_message") as send: - send_to_errors_chat("hello") - send.assert_called_once_with("hello", NOTIFY_LABEL, source="timelock_subscriber", alert_protocol=ALERT_PROTOCOL) + with patch("utils.subscriber.notifier.send_error_message") as send: + send_to_errors_chat("hello", "timelock-subscriber") + send.assert_called_once_with( + "hello", "timelock-subscriber", source="timelock_subscriber", alert_protocol=ALERT_PROTOCOL + ) assert ALERT_PROTOCOL == "yearn-internal" @@ -233,7 +235,7 @@ async def scenario() -> list[str]: server = FakeWSServer(envio) await server.start() sent: list[str] = [] - inbox = EventInbox(str(tmp_path / "sub.db")) + inbox = EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")) notices = PilotNotices( Notifier(send=sent.append, min_interval=0), inbox, TIMELOCKS, NoticeConfig(ws_down_seconds=0.2) ) @@ -306,7 +308,7 @@ def test_baseline_is_persisted_across_restarts(inbox: EventInbox) -> None: def test_fresh_pilot_stays_quiet_about_history_and_restart_announces_downtime_events(tmp_path: Path) -> None: async def run_for(envio: FakeEnvio, server: FakeWSServer, path: str, sent: list[str], body) -> None: # noqa: ANN001 - inbox = EventInbox(path) + inbox = EventInbox(TIMELOCK_STREAM, path) notices = PilotNotices(Notifier(send=sent.append, min_interval=0), inbox, TIMELOCKS) config = WorkerConfig(reconcile_interval=60, heartbeat_interval=0.05, reconcile=ReconcileConfig()) worker = Worker(envio.post, server.url, inbox, SCOPE, config, notices=notices) diff --git a/tests/test_timelock_subscriber_reconcile.py b/tests/test_timelock_subscriber_reconcile.py index 64fee55a..36961b45 100644 --- a/tests/test_timelock_subscriber_reconcile.py +++ b/tests/test_timelock_subscriber_reconcile.py @@ -7,10 +7,11 @@ import pytest -from protocols.timelock.subscriber.envio import EnvioQueryError, fetch_range, range_query -from protocols.timelock.subscriber.inbox import SCHEMA_VERSION, SOURCE_HTTP, SOURCE_WS, EventInbox, event_fingerprint -from protocols.timelock.subscriber.reconcile import ReconcileConfig, build_scope, in_scope, reconcile_all +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.inbox import SCHEMA_VERSION, SOURCE_HTTP, SOURCE_WS, EventInbox, event_fingerprint +from utils.subscriber.reconcile import ReconcileConfig, build_scope, in_scope, reconcile_all SCOPE = {1: {TIMELOCK}, 8453: {TIMELOCK}} CONFIG = ReconcileConfig( @@ -23,7 +24,7 @@ @pytest.fixture def inbox(tmp_path: Path) -> EventInbox: - return EventInbox(str(tmp_path / "sub.db")) + return EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")) def _event_ids(inbox: EventInbox) -> list[str]: @@ -86,13 +87,13 @@ def test_checkpoint_never_moves_backwards(inbox: EventInbox) -> None: def test_scope_filters_address_per_chain() -> None: scope = build_scope([type("T", (), {"address": TIMELOCK.upper().replace("0X", "0x"), "chain_id": 1})()]) - assert in_scope(make_event(1, 1), scope) - assert not in_scope(make_event(8453, 1), scope) - assert not in_scope(make_event(1, 1, address=OTHER_TIMELOCK), scope) + assert in_scope(make_event(1, 1), scope, TIMELOCK_STREAM) + assert not in_scope(make_event(8453, 1), scope, TIMELOCK_STREAM) + assert not in_scope(make_event(1, 1, address=OTHER_TIMELOCK), scope, TIMELOCK_STREAM) def test_range_query_inlines_numbers_and_orders_totally() -> None: - query = range_query(1, 10, 20, 50, 100, since_ts=None) + query = range_query(TIMELOCK_STREAM, 1, 10, 20, 50, 100, since_ts=None) assert "chainId: { _eq: 1 }" in query assert "blockNumber: { _gt: 10, _lte: 20 }" in query assert "order_by: [{ blockNumber: asc }, { logIndex: asc }, { id: asc }]" in query @@ -103,7 +104,7 @@ def test_fetch_range_refuses_unbounded_ranges() -> None: envio = FakeEnvio() envio.add(*(make_event(1, b) for b in range(1, 31))) with pytest.raises(EnvioQueryError, match="exceeds 2 pages"): - fetch_range(envio.post, 1, {TIMELOCK}, 0, 30, page_size=10, max_pages=2) + fetch_range(envio.post, TIMELOCK_STREAM, 1, {TIMELOCK}, 0, 30, page_size=10, max_pages=2) # ── reconcile ────────────────────────────────────────────────────────── @@ -242,14 +243,14 @@ def test_reorg_replaced_and_dropped_events_are_recorded(inbox: EventInbox) -> No def test_restart_resumes_from_persisted_checkpoints(tmp_path: Path) -> None: path = str(tmp_path / "sub.db") envio = FakeEnvio() - _seed(envio, EventInbox(path)) + _seed(envio, EventInbox(TIMELOCK_STREAM, path)) envio.add(make_event(1, 30), make_event(8453, 40)) - reconcile_all(envio.post, EventInbox(path), SCOPE, CONFIG) + reconcile_all(envio.post, EventInbox(TIMELOCK_STREAM, path), SCOPE, CONFIG) # "Restart": a fresh inbox object on the same file, with more events indexed meanwhile. envio.add(make_event(1, 35), make_event(8453, 45)) envio.queries.clear() - restarted = EventInbox(path) + restarted = EventInbox(TIMELOCK_STREAM, path) report = reconcile_all(envio.post, restarted, SCOPE, CONFIG) assert report.ok froms = {q["chain_id"]: q["from"] for q in envio.queries if q["offset"] == 0} @@ -346,13 +347,16 @@ def _v0_database(path: Path) -> dict: def test_v0_database_is_migrated_in_place(tmp_path: Path) -> None: path = tmp_path / "sub.db" _v0_database(path) - inbox = EventInbox(str(path)) + inbox = EventInbox(TIMELOCK_STREAM, str(path)) with sqlite3.connect(path) as conn: assert conn.execute("PRAGMA user_version").fetchone()[0] == SCHEMA_VERSION - assert "block_hash" in {r[1] for r in conn.execute("PRAGMA table_info(timelock_sub_events)")} + columns = {r[1] for r in conn.execute("PRAGMA table_info(timelock_sub_events)")} + assert "block_hash" in columns + assert "contract_address" in columns and "timelock_address" not in columns (row,) = inbox.events_since(0) assert row["event_id"] == "1_10_0" and row["block_hash"] is None and row["first_seen_at"] == 5.0 + assert row["contract_address"] == TIMELOCK assert _checkpoint(inbox, 1) == 10 # Re-observing it with the wider selection is a duplicate, not a reorg. @@ -364,15 +368,30 @@ def test_v0_database_is_migrated_in_place(tmp_path: Path) -> None: assert rows["1_10_0"]["revision"] == 0 and rows["1_10_0"]["ws_first_seen_at"] == 20.0 # Reopening is a no-op, and a real change is still detected afterwards. - reopened = EventInbox(str(path)) + reopened = EventInbox(TIMELOCK_STREAM, str(path)) replaced = make_event(1, 10, blockHash="0x" + "dd" * 32) assert reopened.observe([replaced], SOURCE_HTTP, 30.0).changed == 1 +def test_v1_database_renames_the_address_column_and_keeps_its_rows(tmp_path: Path) -> None: + """v1 (the deployed pilot) had block_hash already; v2 only renames timelock_address.""" + path = tmp_path / "sub.db" + _v0_database(path) + with sqlite3.connect(path) as conn: + conn.execute("ALTER TABLE timelock_sub_events ADD COLUMN block_hash TEXT") + conn.execute("PRAGMA user_version = 1") + + inbox = EventInbox(TIMELOCK_STREAM, str(path)) + + (row,) = inbox.events_since(0) + assert row["contract_address"] == TIMELOCK and _checkpoint(inbox, 1) == 10 + assert inbox.observe([make_event(1, 11)], SOURCE_HTTP, 20.0).new == 1 + + def test_widened_selection_does_not_hide_a_real_change(tmp_path: Path) -> None: path = tmp_path / "sub.db" _v0_database(path) - inbox = EventInbox(str(path)) + inbox = EventInbox(TIMELOCK_STREAM, str(path)) # New field present AND the tx changed: a reorg, even though blockHash is new. result = inbox.observe([make_event(1, 10, transactionHash="0x" + "ee" * 32)], SOURCE_HTTP, 20.0) assert result.changed == 1 diff --git a/tests/test_timelock_subscriber_worker.py b/tests/test_timelock_subscriber_worker.py index 1c51096b..b7d0f7a8 100644 --- a/tests/test_timelock_subscriber_worker.py +++ b/tests/test_timelock_subscriber_worker.py @@ -18,13 +18,14 @@ from websockets.asyncio.server import serve from websockets.typing import Subprotocol -from protocols.timelock.subscriber.inbox import SOURCE_HTTP, SOURCE_WS, EventInbox -from protocols.timelock.subscriber.reconcile import ReconcileConfig, reconcile_all -from protocols.timelock.subscriber.report import PilotReport, _recommend, build_report, format_report, percentile -from protocols.timelock.subscriber.worker import Worker, WorkerConfig, backoff_delay +from protocols.timelock.event_stream import TIMELOCK_STREAM 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.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 +from utils.subscriber.worker import Worker, WorkerConfig, backoff_delay SCOPE = {1: {TIMELOCK}, 8453: {TIMELOCK}} @@ -139,7 +140,7 @@ async def scenario() -> None: _seed_heads(envio) server = FakeWSServer(envio) await server.start() - inbox = EventInbox(str(tmp_path / "sub.db")) + inbox = EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")) # The timed reconcile runs once at startup and then not again during the test, # so the subscription sees the events first; HTTP confirms them on demand. config = _config(reconcile_interval=60) @@ -176,7 +177,7 @@ async def scenario() -> None: # the WebSocket on reconnect and must come from HTTP reconciliation. server = FakeWSServer(envio, limit=3) await server.start() - inbox = EventInbox(str(tmp_path / "sub.db")) + inbox = EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")) worker = Worker(envio.post, server.url, inbox, SCOPE, _config(reconcile_interval=0.3)) async def body() -> None: @@ -226,19 +227,19 @@ async def scenario() -> None: server = FakeWSServer(envio, limit=2) await server.start() try: - first = Worker(envio.post, server.url, EventInbox(path), SCOPE, _config()) + first = Worker(envio.post, server.url, EventInbox(TIMELOCK_STREAM, path), SCOPE, _config()) async def before() -> None: await _until(lambda: first.ws_connected and first.last_reconcile_at is not None) envio.add(_now_event(1, 10)) - await _until(lambda: "1_10_0" in _rows(EventInbox(path))) + await _until(lambda: "1_10_0" in _rows(EventInbox(TIMELOCK_STREAM, path))) await _run_worker(first, before) # Indexed while no worker runs. envio.add(*(_now_event(1, b) for b in range(11, 21)), _now_event(8453, 7)) - inbox = EventInbox(path) + inbox = EventInbox(TIMELOCK_STREAM, path) second = Worker(envio.post, server.url, inbox, SCOPE, _config()) async def after() -> None: @@ -264,7 +265,7 @@ def test_http_only_transport_needs_no_websocket(tmp_path: Path) -> None: async def scenario() -> None: envio = FakeEnvio() _seed_heads(envio) - inbox = EventInbox(str(tmp_path / "sub.db")) + inbox = EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")) worker = Worker(envio.post, None, inbox, SCOPE, _config(transport="http", reconcile_interval=0.05)) async def body() -> None: @@ -283,7 +284,7 @@ def test_health_reports_stalled_chain(tmp_path: Path) -> None: clock = [1_000.0] envio = FakeEnvio() _seed_heads(envio) - inbox = EventInbox(str(tmp_path / "sub.db")) + inbox = EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")) config = WorkerConfig(transport="http", head_stall_seconds=900) worker = Worker(envio.post, None, inbox, SCOPE, config, now=lambda: clock[0]) @@ -298,7 +299,7 @@ def test_health_reports_stalled_chain(tmp_path: Path) -> None: def test_worker_rejects_bad_transport_config(tmp_path: Path) -> None: - inbox = EventInbox(str(tmp_path / "sub.db")) + inbox = EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")) with pytest.raises(ValueError, match="transport"): Worker(FakeEnvio().post, "ws://x", inbox, SCOPE, WorkerConfig(transport="carrier-pigeon")) with pytest.raises(ValueError, match="WebSocket URL"): @@ -328,7 +329,7 @@ def test_recommendation_rules() -> None: def test_report_counts_ws_only_rows_that_http_scanned_but_did_not_return(tmp_path: Path) -> None: - inbox = EventInbox(str(tmp_path / "sub.db")) + inbox = EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")) now = time.time() inbox.observe([make_event(1, 10, ts=int(now) - 10)], SOURCE_WS, now) inbox.observe([make_event(1, 99, ts=int(now) - 5)], SOURCE_WS, now) @@ -369,7 +370,7 @@ def flaky_post(query: str, variables: dict) -> dict: raise sqlite3.IntegrityError("simulated bug") # not an EnvioQueryError return envio.post(query, variables) - inbox = EventInbox(str(tmp_path / "sub.db")) + inbox = EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")) worker = Worker(flaky_post, None, inbox, SCOPE, _config(transport="http", reconcile_interval=0.05)) async def body() -> None: @@ -388,7 +389,7 @@ def test_watchdog_deadline_covers_a_hung_first_reconcile(tmp_path: Path) -> None worker = Worker( FakeEnvio().post, None, - EventInbox(str(tmp_path / "sub.db")), + EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")), SCOPE, _config(transport="http"), now=lambda: clock[0], @@ -403,7 +404,13 @@ def test_watchdog_deadline_covers_a_hung_first_reconcile(tmp_path: Path) -> None def test_watchdog_withheld_when_a_loop_task_ended(tmp_path: Path) -> None: async def scenario() -> None: - worker = Worker(FakeEnvio().post, None, EventInbox(str(tmp_path / "sub.db")), SCOPE, _config(transport="http")) + worker = Worker( + FakeEnvio().post, + None, + EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")), + SCOPE, + _config(transport="http"), + ) worker.started_at = worker.now() async def crashed() -> None: @@ -418,7 +425,7 @@ async def crashed() -> None: def test_report_includes_outage_that_began_before_the_window(tmp_path: Path) -> None: - inbox = EventInbox(str(tmp_path / "sub.db")) + inbox = EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")) inbox.log("worker_started", 0.0, transport="both") inbox.log("ws_connected", 10.0) inbox.log("ws_disconnected", 500.0, error="ConnectionClosedError: boom") @@ -438,7 +445,7 @@ def test_report_includes_outage_that_began_before_the_window(tmp_path: Path) -> def test_report_ignores_closed_outage_before_the_window(tmp_path: Path) -> None: - inbox = EventInbox(str(tmp_path / "sub.db")) + inbox = EventInbox(TIMELOCK_STREAM, str(tmp_path / "sub.db")) inbox.log("worker_started", 0.0, transport="both") inbox.log("ws_connected", 10.0) report = build_report(inbox, hours=1, now=7_200.0) diff --git a/tests/timelock_subscriber_fakes.py b/tests/timelock_subscriber_fakes.py index ed342dd0..1fda8eb6 100644 --- a/tests/timelock_subscriber_fakes.py +++ b/tests/timelock_subscriber_fakes.py @@ -9,10 +9,19 @@ from websockets.asyncio.server import Server, ServerConnection, serve from websockets.typing import Subprotocol -from protocols.timelock.subscriber.envio import EnvioQueryError +from utils.subscriber.envio import EnvioQueryError TIMELOCK = "0x88ba032be87d5ef1fbe87336b7090767f367bf73" OTHER_TIMELOCK = "0xd8236031d8279d82e615af2bfab5fc0127a329ab" +MORPHO_VAULT = "0x68aea7b82df6ccdf76235d46445ed83f85f845a3" # Yearn USDC V1, mainnet + +# Envio entity -> field holding the monitored contract. +ADDRESS_FIELDS = {"TimelockEvent": "timelockAddress", "MorphoGovernanceEvent": "vaultAddress"} + + +def entity_of(row: dict[str, Any]) -> str: + """Return the Envio entity a fake row belongs to.""" + return next(entity for entity, field in ADDRESS_FIELDS.items() if field in row) def make_event( @@ -47,13 +56,41 @@ def make_event( return row +def make_morpho_event( + chain_id: int, + block: int, + log_index: int = 0, + ts: int = 1_700_000_000, + address: str = MORPHO_VAULT, + **extra: Any, +) -> dict[str, Any]: + """Return a MorphoGovernanceEvent row shaped like Envio's response.""" + row: dict[str, Any] = { + "id": f"{chain_id}_{block}_{log_index}", + "vaultAddress": address, + "vaultVersion": 1, + "eventName": "SubmitMarketRemoval", + "chainId": chain_id, + "blockNumber": block, + "blockHash": "0x" + f"{block:064x}", + "blockTimestamp": ts, + "logIndex": log_index, + "transactionHash": "0x" + f"{block:064x}", + "transactionIndex": 0, + "transactionFrom": "0x" + "cd" * 20, + "argsJson": '{"caller":"0x' + "cd" * 20 + '","id":"0x' + "ab" * 32 + '"}', + } + row.update(extra) + return row + + def _int(pattern: str, text: str) -> int | None: match = re.search(pattern, text) return int(match.group(1)) if match else None class FakeEnvio: - """Serves ``chain_metadata`` and ``TimelockEvent`` range pages from in-memory rows.""" + """Serves ``chain_metadata`` and range pages of every entity in ``ADDRESS_FIELDS`` from in-memory rows.""" def __init__(self) -> None: self.rows: dict[str, dict[str, Any]] = {} @@ -63,6 +100,7 @@ def __init__(self) -> None: # chain_id -> remaining failures for range queries on that chain self.fail_chain: dict[int, int] = {} self.fail_metadata = 0 + self.metadata_reads = 0 # Rebound to the WebSocket server's event loop by FakeWSServer.start(). self.changed = asyncio.Event() self.loop: asyncio.AbstractEventLoop | None = None @@ -91,6 +129,7 @@ def _notify(self) -> None: def post(self, query: str, variables: dict[str, Any]) -> dict[str, Any]: with self.lock: if "chain_metadata" in query: + self.metadata_reads += 1 if self.fail_metadata: self.fail_metadata -= 1 raise EnvioQueryError("metadata down") @@ -102,6 +141,8 @@ def post(self, query: str, variables: dict[str, Any]) -> dict[str, Any]: ] } } + entity = next(e for e in ADDRESS_FIELDS if f"{e}(" in query) + address_field = ADDRESS_FIELDS[entity] chain_id = _int(r"chainId: \{ _eq: (\d+) \}", query) assert chain_id is not None if self.fail_chain.get(chain_id): @@ -113,25 +154,30 @@ def post(self, query: str, variables: dict[str, Any]) -> dict[str, Any]: limit = _int(r"limit: (\d+)", query) offset = _int(r"offset: (\d+)", query) assert lo is not None and hi is not None and limit is not None and offset is not None - self.queries.append({"chain_id": chain_id, "from": lo, "to": hi, "offset": offset, "since": since}) + self.queries.append( + {"entity": entity, "chain_id": chain_id, "from": lo, "to": hi, "offset": offset, "since": since} + ) addresses = set(variables["addresses"]) matched = sorted( ( r for r in self.rows.values() - if r["chainId"] == chain_id - and r["timelockAddress"] in addresses + if entity_of(r) == entity + and r["chainId"] == chain_id + and r[address_field] in addresses and lo < r["blockNumber"] <= hi and (since is None or r["blockTimestamp"] >= since) ), key=lambda r: (r["blockNumber"], r["logIndex"], r["id"]), ) - return {"data": {"TimelockEvent": [dict(r) for r in matched[offset : offset + limit]]}} + return {"data": {entity: [dict(r) for r in matched[offset : offset + limit]]}} - def snapshot(self, limit: int) -> list[dict[str, Any]]: + def snapshot(self, limit: int, entity: str = "TimelockEvent") -> list[dict[str, Any]]: with self.lock: rows = sorted( - self.rows.values(), key=lambda r: (r["blockTimestamp"], r["blockNumber"], r["logIndex"]), reverse=True + (r for r in self.rows.values() if entity_of(r) == entity), + key=lambda r: (r["blockTimestamp"], r["blockNumber"], r["logIndex"]), + reverse=True, ) return [dict(r) for r in rows[:limit]] @@ -183,16 +229,18 @@ async def _handle(self, conn: ServerConnection) -> None: while sub["type"] == "pong": sub = json.loads(await conn.recv()) assert sub["type"] == "subscribe" + # A bare test query names no entity: serve timelock rows. + entity = next((e for e in ADDRESS_FIELDS if f"{e}(" in sub["payload"]["query"]), "TimelockEvent") self.subscriptions += 1 if self.ping_first: await conn.send(json.dumps({"type": "ping"})) sent = None while True: self.envio.changed.clear() - snapshot = self.envio.snapshot(self.limit) + snapshot = self.envio.snapshot(self.limit, entity) if snapshot != sent: await conn.send( - json.dumps({"id": sub["id"], "type": "next", "payload": {"data": {"TimelockEvent": snapshot}}}) + json.dumps({"id": sub["id"], "type": "next", "payload": {"data": {entity: snapshot}}}) ) sent = snapshot waiter = asyncio.create_task(self.envio.changed.wait()) diff --git a/utils/http_client.py b/utils/http_client.py index 6e4854fb..170ae3cb 100644 --- a/utils/http_client.py +++ b/utils/http_client.py @@ -18,6 +18,7 @@ def request_with_retry( backoff_factor: float | None = None, timeout: int | None = None, log_errors: bool = True, + session: requests.Session | None = None, **kwargs: Any, ) -> requests.Response: """Make an HTTP request with exponential backoff retry on transient errors. @@ -32,6 +33,8 @@ def request_with_retry( backoff_factor: Multiplier for exponential backoff. Defaults to Config.get_backoff_factor(). timeout: Request timeout in seconds. Defaults to Config.get_request_timeout(). log_errors: Log terminal failures. Disable when the caller reports its own failure. + session: Send through this session to reuse its pooled connections. A session + must not be shared across threads. **kwargs: Additional arguments passed to requests.request(). Returns: @@ -50,7 +53,7 @@ def request_with_retry( last_exception: Exception | None = None for attempt in range(retries + 1): try: - response = requests.request(method, url, timeout=timeout, **kwargs) + response = (session or requests).request(method, url, timeout=timeout, **kwargs) response.raise_for_status() return response except requests.exceptions.HTTPError as e: diff --git a/protocols/timelock/subscriber/envio.py b/utils/subscriber/envio.py similarity index 64% rename from protocols/timelock/subscriber/envio.py rename to utils/subscriber/envio.py index 651cfe81..57a76239 100644 --- a/protocols/timelock/subscriber/envio.py +++ b/utils/subscriber/envio.py @@ -1,6 +1,8 @@ -"""Envio GraphQL queries for the timelock subscriber: chain heads, ranges, and the live subscription.""" +"""Envio GraphQL queries for the event subscriber: chain heads, ranges, and the live subscription.""" import json +import threading +import time from collections.abc import Callable from dataclasses import dataclass from typing import Any @@ -9,34 +11,7 @@ from eth_utils import to_checksum_address from utils.http_client import request_with_retry - -# The hourly monitor's selection plus logIndex and blockHash. Both transports use -# it, so a row fingerprints identically whichever one fetched it, and a reorg that -# replaces the block changes the fingerprint through blockHash. -EVENT_FIELDS = """ - id - timelockAddress - timelockType - eventName - chainId - blockNumber - blockHash - blockTimestamp - logIndex - transactionHash - operationId - index - target - value - data - predecessor - delay - signature - creator - metadata - votesFor - votesAgainst -""" +from utils.subscriber.stream import EventStream CHAIN_METADATA_QUERY = "{ chain_metadata { chain_id latest_processed_block block_height } }" @@ -64,12 +39,26 @@ def lag_blocks(self) -> int: def http_poster(url: str, timeout: int = 30) -> GraphQLPost: - """Return a GraphQL POST function bound to ``url`` with retries on transient errors.""" + """Return a GraphQL POST function bound to ``url`` with retries on transient errors. + + Requests go through one persistent ``requests.Session`` per calling thread, so + every stream's reconcile cycle reuses pooled TCP/TLS connections without + sharing a session across threads. + """ + local = threading.local() def post(query: str, variables: dict[str, Any]) -> dict[str, Any]: + session = getattr(local, "session", None) + if session is None: + session = local.session = requests.Session() try: response = request_with_retry( - "post", url, timeout=timeout, log_errors=False, json={"query": query, "variables": variables} + "post", + url, + timeout=timeout, + log_errors=False, + session=session, + json={"query": query, "variables": variables}, ) return response.json() except (requests.RequestException, ValueError) as exc: @@ -105,7 +94,39 @@ def fetch_chain_heads(post: GraphQLPost) -> dict[int, ChainHead]: return heads -def range_query(chain_id: int, from_block: int, to_block: int, limit: int, offset: int, since_ts: int | None) -> str: +class SharedChainHeads: + """``fetch_chain_heads`` shared by every stream's reconcile cycle. + + Streams reconcile on the same cadence, so a read younger than ``max_age`` + seconds is reused instead of querying ``chain_metadata`` once per stream. A + slightly old head only means a cycle stops a few blocks short; the next one + catches up. + """ + + def __init__(self, post: GraphQLPost, max_age: float = 10.0, clock: Callable[[], float] = time.monotonic) -> None: + self.post = post + self.max_age = max_age + self.clock = clock + self._lock = threading.Lock() + self._heads: dict[int, ChainHead] | None = None + self._fetched_at = 0.0 + + def __call__(self) -> dict[int, ChainHead]: + """Return chain heads, fetching them when the cached read is too old. + + Raises: + EnvioQueryError: The fetch failed; the next call retries. + """ + with self._lock: + if self._heads is None or self.clock() - self._fetched_at > self.max_age: + self._heads = fetch_chain_heads(self.post) + self._fetched_at = self.clock() + return self._heads + + +def range_query( + stream: EventStream, chain_id: int, from_block: int, to_block: int, limit: int, offset: int, since_ts: int | None +) -> str: """Build a page query for one chain's events in the block range ``(from_block, to_block]``. Numbers are inlined rather than passed as ``Int!`` variables: Hasura rejects an @@ -115,23 +136,24 @@ def range_query(chain_id: int, from_block: int, to_block: int, limit: int, offse """ since = f", blockTimestamp: {{ _gte: {int(since_ts)} }}" if since_ts is not None else "" return f""" - query TimelockEventRange($addresses: [String!]!) {{ - TimelockEvent( + query {stream.entity}Range($addresses: [String!]!) {{ + {stream.entity}( where: {{ chainId: {{ _eq: {int(chain_id)} }} - timelockAddress: {{ _in: $addresses }} + {stream.address_field}: {{ _in: $addresses }} blockNumber: {{ _gt: {int(from_block)}, _lte: {int(to_block)} }}{since} }} order_by: [{{ blockNumber: asc }}, {{ logIndex: asc }}, {{ id: asc }}] limit: {int(limit)} offset: {int(offset)} - ) {{{EVENT_FIELDS} }} + ) {{{stream.fields} }} }} """ def fetch_range( post: GraphQLPost, + stream: EventStream, chain_id: int, addresses: set[str], from_block: int, @@ -152,8 +174,8 @@ def fetch_range( variables = {"addresses": address_variants(addresses)} rows: list[dict[str, Any]] = [] for page in range(max_pages): - query = range_query(chain_id, from_block, to_block, page_size, page * page_size, since_ts) - batch = _data(post(query, variables)).get("TimelockEvent") or [] + 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 [] rows.extend(batch) if len(batch) < page_size: return rows, page + 1 @@ -163,7 +185,7 @@ def fetch_range( ) -def live_subscription(limit: int, chain_ids: list[int]) -> str: +def live_subscription(stream: EventStream, limit: int, chain_ids: list[int]) -> str: """Live query over the newest monitored events across chains. Hasura re-sends the full snapshot whenever the result changes, so every message @@ -173,11 +195,11 @@ def live_subscription(limit: int, chain_ids: list[int]) -> str: """ chains = ", ".join(str(int(c)) for c in sorted(chain_ids)) return f""" - subscription TimelockEventLive($addresses: [String!]!) {{ - TimelockEvent( - where: {{ timelockAddress: {{ _in: $addresses }}, chainId: {{ _in: [{chains}] }} }} + subscription {stream.entity}Live($addresses: [String!]!) {{ + {stream.entity}( + where: {{ {stream.address_field}: {{ _in: $addresses }}, chainId: {{ _in: [{chains}] }} }} order_by: [{{ blockTimestamp: desc }}, {{ blockNumber: desc }}, {{ logIndex: desc }}] limit: {int(limit)} - ) {{{EVENT_FIELDS} }} + ) {{{stream.fields} }} }} """ diff --git a/protocols/timelock/subscriber/inbox.py b/utils/subscriber/inbox.py similarity index 76% rename from protocols/timelock/subscriber/inbox.py rename to utils/subscriber/inbox.py index 41bbc251..4275e2f5 100644 --- a/protocols/timelock/subscriber/inbox.py +++ b/utils/subscriber/inbox.py @@ -1,16 +1,18 @@ -"""Durable local inbox for timelock events seen by the continuous subscriber. - -Every ``TimelockEvent`` row the worker sees, from the WebSocket subscription or -from HTTP reconciliation, is folded into one row per Envio event id -(``__``) with its provenance: which transport saw -it first, when each transport first saw it, how often it was re-observed, and a -fingerprint of its content, including the block hash. A row that comes back -with different content (a reorg replaced the block or the log at that position) or disappears from a reconciled range -(a reorg dropped it) is recorded in ``timelock_sub_changes`` instead of silently -overwritten. - -Per-chain checkpoints live here too, independent of the hourly monitor's -``TIMELOCK_LAST_TS``. The pilot writes nothing else: no Telegram, no alert store. +"""Durable local inbox for one event stream seen by the continuous subscriber. + +Every row of the stream's Envio entity the worker sees, from the WebSocket +subscription or from HTTP reconciliation, is folded into one row per Envio event +id (``__``) with its provenance: which transport +saw it first, when each transport first saw it, how often it was re-observed, +and a fingerprint of its content, including the block hash. A row that comes +back with different content (a reorg replaced the block or the log at that +position) or disappears from a reconciled range (a reorg dropped it) is recorded +in ``_changes`` instead of silently overwritten. + +Each stream has its own SQLite file and table prefix (see ``EventStream``), so +streams never contend for one write lock. Per-chain checkpoints live here too, +independent of the hourly monitors' cursors. The inbox writes nothing else: no +Telegram, no alert store. """ import hashlib @@ -22,17 +24,16 @@ from typing import Any from utils.paths import cache_path - -DB_FILENAME = "timelock_subscriber.db" +from utils.subscriber.stream import EventStream SOURCE_WS = "ws" SOURCE_HTTP = "http" -_SCHEMA = """ -CREATE TABLE IF NOT EXISTS timelock_sub_events ( +_SCHEMA_TEMPLATE = """ +CREATE TABLE IF NOT EXISTS {p}_events ( event_id TEXT PRIMARY KEY, chain_id INTEGER NOT NULL, - timelock_address TEXT NOT NULL, + contract_address TEXT NOT NULL, block_number INTEGER NOT NULL, block_hash TEXT, log_index INTEGER NOT NULL, @@ -52,12 +53,12 @@ http_observations INTEGER NOT NULL DEFAULT 0, updated_at REAL NOT NULL ); -CREATE INDEX IF NOT EXISTS idx_timelock_sub_events_chain_block - ON timelock_sub_events(chain_id, block_number); -CREATE INDEX IF NOT EXISTS idx_timelock_sub_events_first_seen - ON timelock_sub_events(first_seen_at); +CREATE INDEX IF NOT EXISTS idx_{p}_events_chain_block + ON {p}_events(chain_id, block_number); +CREATE INDEX IF NOT EXISTS idx_{p}_events_first_seen + ON {p}_events(first_seen_at); -CREATE TABLE IF NOT EXISTS timelock_sub_changes ( +CREATE TABLE IF NOT EXISTS {p}_changes ( id INTEGER PRIMARY KEY AUTOINCREMENT, event_id TEXT NOT NULL, chain_id INTEGER NOT NULL, @@ -69,7 +70,7 @@ old_payload_json TEXT ); -CREATE TABLE IF NOT EXISTS timelock_sub_checkpoints ( +CREATE TABLE IF NOT EXISTS {p}_checkpoints ( chain_id INTEGER PRIMARY KEY, block_number INTEGER NOT NULL, indexed_head INTEGER NOT NULL, @@ -77,16 +78,16 @@ updated_at REAL NOT NULL ); -CREATE TABLE IF NOT EXISTS timelock_sub_log ( +CREATE TABLE IF NOT EXISTS {p}_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, kind TEXT NOT NULL, at REAL NOT NULL, duration_ms REAL, details_json TEXT ); -CREATE INDEX IF NOT EXISTS idx_timelock_sub_log_kind_at ON timelock_sub_log(kind, at); +CREATE INDEX IF NOT EXISTS idx_{p}_log_kind_at ON {p}_log(kind, at); -CREATE TABLE IF NOT EXISTS timelock_sub_status ( +CREATE TABLE IF NOT EXISTS {p}_status ( key TEXT PRIMARY KEY, value TEXT NOT NULL, updated_at REAL NOT NULL @@ -96,18 +97,26 @@ # Current schema version (PRAGMA user_version). Databases from older revisions are # migrated in place by _migrate; observations, changes, and checkpoints are kept. -SCHEMA_VERSION = 1 +SCHEMA_VERSION = 2 + + +def _schema(stream: EventStream) -> str: + """Return the CREATE statements for ``stream``'s tables.""" + return _SCHEMA_TEMPLATE.replace("{p}", stream.table_prefix) def _columns(conn: sqlite3.Connection, table: str) -> set[str]: + """Return the column names of ``table``.""" return {row[1] for row in conn.execute(f"PRAGMA table_info({table})")} -def _migrate(conn: sqlite3.Connection) -> None: +def _migrate(conn: sqlite3.Connection, events_table: str) -> None: """Bring a database created by an earlier revision up to ``SCHEMA_VERSION``. - v0 → v1: ``timelock_sub_events.block_hash`` (rows observed before the event + v0 → v1: ``_events.block_hash`` (rows observed before the event selection included ``blockHash`` keep NULL until they are next observed). + v1 → v2: ``timelock_address`` renamed to ``contract_address`` once the inbox + served more than timelocks. """ if conn.execute("PRAGMA user_version").fetchone()[0] >= SCHEMA_VERSION: return @@ -116,8 +125,11 @@ def _migrate(conn: sqlite3.Connection) -> None: conn.execute("BEGIN IMMEDIATE") if conn.execute("PRAGMA user_version").fetchone()[0] >= SCHEMA_VERSION: return - if "block_hash" not in _columns(conn, "timelock_sub_events"): - conn.execute("ALTER TABLE timelock_sub_events ADD COLUMN block_hash TEXT") + columns = _columns(conn, events_table) + if "block_hash" not in columns: + conn.execute(f"ALTER TABLE {events_table} ADD COLUMN block_hash TEXT") + if "timelock_address" in columns: + conn.execute(f"ALTER TABLE {events_table} RENAME COLUMN timelock_address TO contract_address") conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}") @@ -177,6 +189,7 @@ def event_fingerprint(event: dict[str, Any]) -> str: def _lower(value: Any) -> str | None: + """Lowercase ``value`` as a string, keeping None.""" return str(value).lower() if value is not None else None @@ -188,17 +201,29 @@ def parse_log_index(event: dict[str, Any]) -> int: class EventInbox: - """SQLite-backed store for observed events, changes, checkpoints, and worker log. + """SQLite-backed store for one stream's observed events, changes, checkpoints, and worker log. Each call opens its own short-lived connection so the WebSocket task and the reconcile thread can share the database safely (WAL mode, busy timeout). """ - def __init__(self, path: str | None = None) -> None: - self.path = path or cache_path(DB_FILENAME) + def __init__(self, stream: EventStream, path: str | None = None) -> None: + self.stream = stream + self.path = path or cache_path(stream.db_filename) + self._events = stream.table("events") + self._changes = stream.table("changes") + self._checkpoints = stream.table("checkpoints") + self._log = stream.table("log") + self._status = stream.table("status") + self._schema = _schema(stream) with self._connect(): pass + def _operation_id(self, event: dict[str, Any]) -> Any: + """Return the stream's operation id field of ``event``, if it has one.""" + field = self.stream.operation_field + return event.get(field) if field else None + @contextmanager def _connect(self, write: bool = False) -> Iterator[sqlite3.Connection]: """Open a connection; one transaction spans the ``with`` block. @@ -214,8 +239,8 @@ def _connect(self, write: bool = False) -> Iterator[sqlite3.Connection]: conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA synchronous=NORMAL") conn.execute("PRAGMA busy_timeout=10000") - conn.executescript(_SCHEMA) - _migrate(conn) + conn.executescript(self._schema) + _migrate(conn, self._events) with conn: if write: conn.execute("BEGIN IMMEDIATE") @@ -227,10 +252,10 @@ def observe(self, events: Iterable[dict[str, Any]], source: str, now: float) -> A known id with the same fingerprint only bumps the per-source counters (a duplicate snapshot or an overlapping reconcile). A known id with a different fingerprint is a changed occurrence; a previously removed id coming back is - restored. Both are logged in ``timelock_sub_changes``. + restored. Both are logged in ``_changes``. Args: - events: ``TimelockEvent`` rows as returned by Envio. + events: Rows of the stream's entity as returned by Envio. source: ``SOURCE_WS`` or ``SOURCE_HTTP``. now: Observation time (unix seconds). @@ -247,14 +272,14 @@ def observe(self, events: Iterable[dict[str, Any]], source: str, now: float) -> event_id = str(event["id"]) fingerprint = event_fingerprint(event) row = conn.execute( - "SELECT fingerprint, status, payload_json FROM timelock_sub_events WHERE event_id = ?", + f"SELECT fingerprint, status, payload_json FROM {self._events} WHERE event_id = ?", (event_id,), ).fetchone() if row is None: conn.execute( f""" - INSERT INTO timelock_sub_events ( - event_id, chain_id, timelock_address, block_number, block_hash, log_index, + INSERT INTO {self._events} ( + event_id, chain_id, contract_address, block_number, block_hash, log_index, block_timestamp, transaction_hash, event_name, operation_id, fingerprint, payload_json, first_seen_at, first_source, {seen_col}, {count_col}, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?) @@ -262,14 +287,14 @@ def observe(self, events: Iterable[dict[str, Any]], source: str, now: float) -> ( event_id, int(event["chainId"]), - str(event["timelockAddress"]).lower(), + str(event[self.stream.address_field]).lower(), int(event["blockNumber"]), _lower(event.get("blockHash")), parse_log_index(event), int(event["blockTimestamp"]), str(event["transactionHash"]).lower(), event.get("eventName"), - event.get("operationId"), + self._operation_id(event), fingerprint, json.dumps(event, sort_keys=True, default=str), now, @@ -291,7 +316,7 @@ def observe(self, events: Iterable[dict[str, Any]], source: str, now: float) -> ): # Same event, stored before the selection grew: upgrade it in place. conn.execute( - "UPDATE timelock_sub_events SET fingerprint = ?, payload_json = ?, block_hash = ? " + f"UPDATE {self._events} SET fingerprint = ?, payload_json = ?, block_hash = ? " "WHERE event_id = ?", ( fingerprint, @@ -310,8 +335,8 @@ def observe(self, events: Iterable[dict[str, Any]], source: str, now: float) -> if kind: changed_ids.append(event_id) conn.execute( - """ - INSERT INTO timelock_sub_changes ( + f""" + INSERT INTO {self._changes} ( event_id, chain_id, kind, old_fingerprint, new_fingerprint, source, observed_at, old_payload_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) @@ -328,8 +353,8 @@ def observe(self, events: Iterable[dict[str, Any]], source: str, now: float) -> ), ) conn.execute( - """ - UPDATE timelock_sub_events SET + f""" + UPDATE {self._events} SET status = 'active', revision = revision + 1, fingerprint = ?, payload_json = ?, block_hash = ?, block_timestamp = ?, transaction_hash = ?, event_name = ?, operation_id = ? @@ -342,13 +367,13 @@ def observe(self, events: Iterable[dict[str, Any]], source: str, now: float) -> int(event["blockTimestamp"]), str(event["transactionHash"]).lower(), event.get("eventName"), - event.get("operationId"), + self._operation_id(event), event_id, ), ) conn.execute( f""" - UPDATE timelock_sub_events SET + UPDATE {self._events} SET {seen_col} = COALESCE({seen_col}, ?), {count_col} = {count_col} + 1, updated_at = ? WHERE event_id = ? """, @@ -371,8 +396,8 @@ def mark_missing( """ with self._connect(write=True) as conn: rows = conn.execute( - """ - SELECT event_id, fingerprint, payload_json FROM timelock_sub_events + f""" + SELECT event_id, fingerprint, payload_json FROM {self._events} WHERE chain_id = ? AND block_number BETWEEN ? AND ? AND status = 'active' """, (chain_id, from_block, to_block), @@ -380,8 +405,8 @@ def mark_missing( removed = [row for row in rows if row["event_id"] not in seen_ids] for row in removed: conn.execute( - """ - INSERT INTO timelock_sub_changes ( + f""" + INSERT INTO {self._changes} ( event_id, chain_id, kind, old_fingerprint, new_fingerprint, source, observed_at, old_payload_json ) VALUES (?, ?, 'removed', ?, NULL, ?, ?, ?) @@ -389,7 +414,7 @@ def mark_missing( (row["event_id"], chain_id, row["fingerprint"], source, now, row["payload_json"]), ) conn.execute( - "UPDATE timelock_sub_events SET status = 'removed', revision = revision + 1, updated_at = ? " + f"UPDATE {self._events} SET status = 'removed', revision = revision + 1, updated_at = ? " "WHERE event_id = ?", (now, row["event_id"]), ) @@ -398,7 +423,7 @@ def mark_missing( def get_checkpoint(self, chain_id: int) -> Checkpoint | None: """Return the stored checkpoint for a chain, or None before its first reconcile.""" with self._connect() as conn: - row = conn.execute("SELECT * FROM timelock_sub_checkpoints WHERE chain_id = ?", (chain_id,)).fetchone() + row = conn.execute(f"SELECT * FROM {self._checkpoints} WHERE chain_id = ?", (chain_id,)).fetchone() if row is None: return None return Checkpoint( @@ -412,14 +437,14 @@ def get_checkpoint(self, chain_id: int) -> Checkpoint | None: def checkpoints(self) -> list[Checkpoint]: """Return every chain's checkpoint, ordered by chain id.""" with self._connect() as conn: - ids = [r["chain_id"] for r in conn.execute("SELECT chain_id FROM timelock_sub_checkpoints ORDER BY 1")] + ids = [r["chain_id"] for r in conn.execute(f"SELECT chain_id FROM {self._checkpoints} ORDER BY 1")] return [cp for chain_id in ids if (cp := self.get_checkpoint(chain_id))] def set_checkpoint(self, chain_id: int, block_number: int, indexed_head: int, now: float) -> None: """Advance a chain's checkpoint; it never moves backwards.""" with self._connect(write=True) as conn: previous = conn.execute( - "SELECT block_number, indexed_head, head_changed_at FROM timelock_sub_checkpoints WHERE chain_id = ?", + f"SELECT block_number, indexed_head, head_changed_at FROM {self._checkpoints} WHERE chain_id = ?", (chain_id,), ).fetchone() head_changed_at = now @@ -429,8 +454,8 @@ def set_checkpoint(self, chain_id: int, block_number: int, indexed_head: int, no indexed_head = previous["indexed_head"] head_changed_at = previous["head_changed_at"] conn.execute( - """ - INSERT INTO timelock_sub_checkpoints (chain_id, block_number, indexed_head, head_changed_at, updated_at) + f""" + INSERT INTO {self._checkpoints} (chain_id, block_number, indexed_head, head_changed_at, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(chain_id) DO UPDATE SET block_number = excluded.block_number, indexed_head = excluded.indexed_head, @@ -443,14 +468,14 @@ def log(self, kind: str, now: float, duration_ms: float | None = None, **details """Append a worker log entry (ws_connected, ws_disconnected, reconcile, ...).""" with self._connect() as conn: conn.execute( - "INSERT INTO timelock_sub_log (kind, at, duration_ms, details_json) VALUES (?, ?, ?, ?)", + f"INSERT INTO {self._log} (kind, at, duration_ms, details_json) VALUES (?, ?, ?, ?)", (kind, now, duration_ms, json.dumps(details, sort_keys=True, default=str)), ) def log_entries(self, since: float, kinds: Iterable[str] | None = None) -> list[sqlite3.Row]: """Return worker log entries at or after ``since``, oldest first.""" with self._connect() as conn: - rows = conn.execute("SELECT * FROM timelock_sub_log WHERE at >= ? ORDER BY at, id", (since,)).fetchall() + rows = conn.execute(f"SELECT * FROM {self._log} WHERE at >= ? ORDER BY at, id", (since,)).fetchall() wanted = set(kinds) if kinds is not None else None return [row for row in rows if wanted is None or row["kind"] in wanted] @@ -460,7 +485,7 @@ def last_log_entry_before(self, before: float, kinds: Iterable[str]) -> sqlite3. placeholders = ", ".join("?" for _ in wanted) with self._connect() as conn: return conn.execute( - f"SELECT * FROM timelock_sub_log WHERE at < ? AND kind IN ({placeholders}) ORDER BY at DESC, id DESC " + f"SELECT * FROM {self._log} WHERE at < ? AND kind IN ({placeholders}) ORDER BY at DESC, id DESC " "LIMIT 1", (before, *wanted), ).fetchone() @@ -472,7 +497,7 @@ def get_events(self, event_ids: Iterable[str]) -> list[sqlite3.Row]: return [] placeholders = ", ".join("?" for _ in ids) with self._connect() as conn: - rows = conn.execute(f"SELECT * FROM timelock_sub_events WHERE event_id IN ({placeholders})", ids).fetchall() + rows = conn.execute(f"SELECT * FROM {self._events} WHERE event_id IN ({placeholders})", ids).fetchall() by_id = {row["event_id"]: row for row in rows} return [by_id[event_id] for event_id in ids if event_id in by_id] @@ -480,8 +505,7 @@ def events_since(self, since_block_timestamp: int) -> list[sqlite3.Row]: """Return inbox rows whose block timestamp is at or after ``since_block_timestamp``.""" with self._connect() as conn: return conn.execute( - "SELECT * FROM timelock_sub_events WHERE block_timestamp >= ? ORDER BY chain_id, block_number, " - "log_index", + f"SELECT * FROM {self._events} WHERE block_timestamp >= ? ORDER BY chain_id, block_number, log_index", (since_block_timestamp,), ).fetchall() @@ -489,15 +513,15 @@ def changes_since(self, since: float) -> list[sqlite3.Row]: """Return change records observed at or after ``since``.""" with self._connect() as conn: return conn.execute( - "SELECT * FROM timelock_sub_changes WHERE observed_at >= ? ORDER BY observed_at, id", (since,) + f"SELECT * FROM {self._changes} WHERE observed_at >= ? ORDER BY observed_at, id", (since,) ).fetchall() def set_status(self, key: str, value: Any, now: float) -> None: """Store one worker status value (JSON-encoded).""" with self._connect() as conn: conn.execute( - """ - INSERT INTO timelock_sub_status (key, value, updated_at) VALUES (?, ?, ?) + f""" + INSERT INTO {self._status} (key, value, updated_at) VALUES (?, ?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at """, (key, json.dumps(value, sort_keys=True, default=str), now), @@ -506,5 +530,5 @@ def set_status(self, key: str, value: Any, now: float) -> None: def status(self) -> dict[str, tuple[Any, float]]: """Return all worker status values with their update time.""" with self._connect() as conn: - rows = conn.execute("SELECT key, value, updated_at FROM timelock_sub_status").fetchall() + rows = conn.execute(f"SELECT key, value, updated_at FROM {self._status}").fetchall() return {row["key"]: (json.loads(row["value"]), row["updated_at"]) for row in rows} diff --git a/protocols/timelock/subscriber/messages.py b/utils/subscriber/messages.py similarity index 71% rename from protocols/timelock/subscriber/messages.py rename to utils/subscriber/messages.py index 28322a51..095c03ae 100644 --- a/protocols/timelock/subscriber/messages.py +++ b/utils/subscriber/messages.py @@ -1,7 +1,8 @@ """Plain-text notices the subscriber pilot posts to the internal errors chat. -They are diagnostics for comparing the pilot with the hourly monitor, not alerts: -short, plain text, with full addresses and explorer links. +They are diagnostics for comparing the pilot with the scheduled monitors, not +alerts: short, plain text, with full addresses and explorer links. Text that +names the event kind uses the stream's noun ("timelock", "Morpho governance"). """ import json @@ -9,9 +10,10 @@ from collections.abc import Mapping, Sequence from typing import Any -from protocols.timelock.subscriber.report import LatencyStats, PilotReport from utils.chains import EXPLORER_URLS, Chain from utils.formatting import format_duration +from utils.subscriber.report import LatencyStats, PilotReport +from utils.subscriber.stream import EventStream def chain_name(chain_id: int) -> str: @@ -23,40 +25,45 @@ def chain_name(chain_id: int) -> str: def _address_line(label: str, address: str, chain_id: int) -> str: + """Render ``label: address`` with an explorer link when the chain has one.""" explorer = EXPLORER_URLS.get(chain_id) return f"{label}: {address}" + (f" {explorer}/address/{address}" if explorer else "") -def format_new_event(row: sqlite3.Row, timelocks: Mapping[tuple[str, int], Any]) -> str: +def format_new_event(row: sqlite3.Row, labels: Mapping[tuple[str, int], Any], stream: EventStream) -> str: """Describe one newly observed event: what, where, which transport, and how late. Args: - row: An inbox row (``timelock_sub_events``). - timelocks: ``(lowercase address, chain id)`` → config with ``protocol`` and ``label``. + row: An inbox row (``_events``). + labels: ``(lowercase address, chain id)`` → object with ``protocol`` and ``label``. + stream: The stream the row belongs to. """ chain_id = int(row["chain_id"]) - address = str(row["timelock_address"]) - config = timelocks.get((address.lower(), chain_id)) - name = f"{config.label} ({config.protocol})" if config else "Unknown timelock" + address = str(row["contract_address"]) + config = labels.get((address.lower(), chain_id)) + name = f"{config.label} ({config.protocol})" if config else f"Unknown {stream.address_label.lower()}" payload = json.loads(row["payload_json"]) delay = row["first_seen_at"] - row["block_timestamp"] explorer = EXPLORER_URLS.get(chain_id) tx = str(payload.get("transactionHash") or row["transaction_hash"]) lines = [ - f"🛰️ New timelock event via {str(row['first_source']).upper()}, {format_duration(int(delay))} after its block", + f"🛰️ New {stream.noun} event via {str(row['first_source']).upper()}, " + f"{format_duration(int(delay))} after its block", f"{name} on {chain_name(chain_id)}: {row['event_name']} {row['event_id']}", - _address_line("Timelock", address, chain_id), + _address_line(stream.address_label, address, chain_id), f"Tx: {tx}" + (f" {explorer}/tx/{tx}" if explorer else ""), ] if row["operation_id"]: lines.append(f"Operation ID: {row['operation_id']}") + if stream.describe is not None: + lines.extend(stream.describe(payload)) return "\n".join(lines) -def format_event_burst(chain_id: int, source: str, event_ids: Sequence[str]) -> str: +def format_event_burst(chain_id: int, source: str, event_ids: Sequence[str], stream: EventStream) -> str: """Summarize many new events at once (a backlog) instead of one notice each.""" shown = ", ".join(event_ids[:20]) + (f" … (+{len(event_ids) - 20})" if len(event_ids) > 20 else "") - return f"🛰️ {len(event_ids)} new timelock events on {chain_name(chain_id)} via {source.upper()}: {shown}" + return f"🛰️ {len(event_ids)} new {stream.noun} events on {chain_name(chain_id)} via {source.upper()}: {shown}" def format_ws_miss(chain_id: int, event_ids: Sequence[str]) -> str: @@ -67,11 +74,11 @@ def format_ws_miss(chain_id: int, event_ids: Sequence[str]) -> str: ) -def format_changes(chain_id: int, source: str, changed: int, restored: int, removed: int) -> str: +def format_changes(chain_id: int, source: str, changed: int, restored: int, removed: int, stream: EventStream) -> str: """Indexed rows changed, vanished, or came back: a reorg or an indexer rewrite.""" return ( - f"🔀 Indexed timelock events changed on {chain_name(chain_id)} (seen via {source.upper()}): " - f"{changed} changed, {removed} removed, {restored} restored. Details: timelock_sub_changes." + f"🔀 Indexed {stream.noun} events changed on {chain_name(chain_id)} (seen via {source.upper()}): " + f"{changed} changed, {removed} removed, {restored} restored. Details: {stream.table('changes')}." ) @@ -106,15 +113,16 @@ def format_reconcile_recovered(failures: int) -> str: def _lat(stats: LatencyStats) -> str: + """Render latency percentiles, or n/a without samples.""" return "n/a" if not stats.count else f"p50 {stats.p50}s / p95 {stats.p95}s (n={stats.count})" -def format_daily_summary(report: PilotReport) -> str: +def format_daily_summary(report: PilotReport, stream: EventStream) -> str: """Condense a pilot report to a few lines.""" coverage = "n/a" if report.ws_coverage is None else f"{report.ws_coverage:.1%}" return "\n".join( [ - "📊 Timelock subscriber, last 24h", + f"📊 {stream.noun[:1].upper()}{stream.noun[1:]} subscriber, last 24h", f"Events {report.events} (HTTP-confirmed {report.confirmed_by_http}), WS coverage {coverage}, " f"WS-only {report.ws_only}, HTTP-only {report.http_only}", f"Delay WS {_lat(report.ws_latency)}; HTTP {_lat(report.http_latency)}; WS lead {_lat(report.ws_lead)}", diff --git a/protocols/timelock/subscriber/notices.py b/utils/subscriber/notices.py similarity index 90% rename from protocols/timelock/subscriber/notices.py rename to utils/subscriber/notices.py index 1fd6dd5c..32043e9c 100644 --- a/protocols/timelock/subscriber/notices.py +++ b/utils/subscriber/notices.py @@ -20,11 +20,11 @@ from dataclasses import dataclass from typing import Any -from protocols.timelock.subscriber import messages -from protocols.timelock.subscriber.inbox import EventInbox -from protocols.timelock.subscriber.notifier import Notifier -from protocols.timelock.subscriber.reconcile import ReconcileReport -from protocols.timelock.subscriber.report import build_report +from utils.subscriber import messages +from utils.subscriber.inbox import EventInbox +from utils.subscriber.notifier import Notifier +from utils.subscriber.reconcile import ReconcileReport +from utils.subscriber.report import build_report SUMMARY_STATUS_KEY = "last_summary_at" # When notices were first enabled on this inbox. Events from blocks before it are @@ -45,23 +45,25 @@ class NoticeConfig: def _chain_of(event_id: str) -> int: + """Return the chain id prefix of an Envio event id.""" return int(event_id.split("_", 1)[0]) class PilotNotices: - """Notice state and policy for one worker run.""" + """Notice state and policy for one stream's worker run.""" def __init__( self, notifier: Notifier, inbox: EventInbox, - timelocks: Mapping[tuple[str, int], Any], + labels: Mapping[tuple[str, int], Any], config: NoticeConfig = NoticeConfig(), now: Callable[[], float] = time.time, ) -> None: self.notifier = notifier self.inbox = inbox - self.timelocks = timelocks + self.stream = inbox.stream + self.labels = labels self.config = config self.ws_down_since: float | None = None self.ws_last_error: str | None = None @@ -107,10 +109,10 @@ async def new_events(self, source: str, event_ids: list[str] | tuple[str, ...]) for event_id in event_ids: by_chain.setdefault(_chain_of(event_id), []).append(event_id) for chain_id, ids in sorted(by_chain.items()): - self.notifier.post(messages.format_event_burst(chain_id, source, ids)) + self.notifier.post(messages.format_event_burst(chain_id, source, ids, self.stream)) return for row in await asyncio.to_thread(self.inbox.get_events, event_ids): - self.notifier.post(messages.format_new_event(row, self.timelocks)) + self.notifier.post(messages.format_new_event(row, self.labels, self.stream)) def snapshot_changes(self, changed_ids: list[str] | tuple[str, ...]) -> None: """Post one notice per chain for rows a WebSocket snapshot showed changed or restored.""" @@ -118,7 +120,9 @@ def snapshot_changes(self, changed_ids: list[str] | tuple[str, ...]) -> None: for event_id in changed_ids: by_chain[_chain_of(event_id)] = by_chain.get(_chain_of(event_id), 0) + 1 for chain_id, count in sorted(by_chain.items()): - self.notifier.post(messages.format_changes(chain_id, "ws", changed=count, restored=0, removed=0)) + self.notifier.post( + messages.format_changes(chain_id, "ws", changed=count, restored=0, removed=0, stream=self.stream) + ) async def reconcile(self, report: ReconcileReport, ws_expected: bool) -> None: """Post notices for one reconcile cycle: new events, WS misses, reorgs, failure streaks.""" @@ -132,7 +136,9 @@ async def reconcile(self, report: ReconcileReport, ws_expected: bool) -> None: self.notifier.post(messages.format_ws_miss(chain.chain_id, missed)) if chain.changed or chain.restored or chain.removed: self.notifier.post( - messages.format_changes(chain.chain_id, "http", chain.changed, chain.restored, chain.removed) + messages.format_changes( + chain.chain_id, "http", chain.changed, chain.restored, chain.removed, self.stream + ) ) if report.ok: @@ -190,5 +196,5 @@ async def maybe_daily_summary(self, now: float) -> None: if now - float(last) < self.config.summary_interval: return report = await asyncio.to_thread(build_report, self.inbox, self.config.summary_interval / 3600, now) - self.notifier.post(messages.format_daily_summary(report)) + self.notifier.post(messages.format_daily_summary(report, self.stream)) await asyncio.to_thread(self.inbox.set_status, SUMMARY_STATUS_KEY, now, now) diff --git a/protocols/timelock/subscriber/notifier.py b/utils/subscriber/notifier.py similarity index 75% rename from protocols/timelock/subscriber/notifier.py rename to utils/subscriber/notifier.py index ae194f47..56b4f73a 100644 --- a/protocols/timelock/subscriber/notifier.py +++ b/utils/subscriber/notifier.py @@ -8,34 +8,38 @@ import asyncio from collections.abc import Callable +from functools import partial from utils.logger import get_logger from utils.telegram import MAX_MESSAGE_LENGTH, error_channel_configured, send_error_message -_logger = get_logger("timelock.subscriber.notifier") +_logger = get_logger("subscriber.notifier") -# Label shown as "[timelock-subscriber]" in the errors chat. -NOTIFY_LABEL = "timelock-subscriber" # Alert-history key: internal-only, so pilot notices stay off public protocol pages. ALERT_PROTOCOL = "yearn-internal" NOTIFY_MODES = ("none", "errors") _SEPARATOR = "\n\n" -# Room for the "[timelock-subscriber] " prefix and Markdown escaping of the text. +# Room for the "[-subscriber] " prefix and Markdown escaping of the text. _MAX_BATCH_CHARS = MAX_MESSAGE_LENGTH // 2 SendFn = Callable[[str], None] -def send_to_errors_chat(text: str) -> None: - """Send one plain-text notice to the internal errors chat, silently.""" - send_error_message(text, NOTIFY_LABEL, source="timelock_subscriber", alert_protocol=ALERT_PROTOCOL) +def notice_label(stream_name: str) -> str: + """Return the errors-chat label for a stream's notices, e.g. ``[timelock-subscriber]``.""" + return f"{stream_name}-subscriber" + + +def send_to_errors_chat(text: str, label: str) -> None: + """Send one plain-text notice to the internal errors chat under ``[label]``, silently.""" + send_error_message(text, label, source=label.replace("-", "_"), alert_protocol=ALERT_PROTOCOL) class Notifier: """Queue of notices drained by one background sender task.""" - def __init__(self, send: SendFn = send_to_errors_chat, max_queue: int = 200, min_interval: float = 1.0) -> None: + def __init__(self, send: SendFn, max_queue: int = 200, min_interval: float = 1.0) -> None: self.send = send self.min_interval = min_interval self.queue: asyncio.Queue[str] = asyncio.Queue(maxsize=max_queue) @@ -81,8 +85,8 @@ async def run(self) -> None: await asyncio.sleep(self.min_interval) -def build_notifier(mode: str) -> Notifier | None: - """Return a notifier for ``mode``, or None when notices are off or cannot be routed safely. +def build_notifier(mode: str, stream_name: str) -> Notifier | None: + """Return a notifier for one stream's notices, or None when notices are off or cannot be routed safely. ``errors`` requires a configured errors chat: ``send_error_message`` would otherwise fall back to a protocol channel, and pilot notices must never reach one. @@ -92,6 +96,6 @@ def build_notifier(mode: str) -> Notifier | None: if mode == "none": return None if not error_channel_configured(): - _logger.warning("TIMELOCK_SUBSCRIBER_NOTIFY=errors but no errors chat is configured; notices disabled") + _logger.warning("subscriber notices set to errors but no errors chat is configured; notices disabled") return None - return Notifier() + return Notifier(send=partial(send_to_errors_chat, label=notice_label(stream_name))) diff --git a/protocols/timelock/subscriber/reconcile.py b/utils/subscriber/reconcile.py similarity index 80% rename from protocols/timelock/subscriber/reconcile.py rename to utils/subscriber/reconcile.py index 18c0db9b..46fcb975 100644 --- a/protocols/timelock/subscriber/reconcile.py +++ b/utils/subscriber/reconcile.py @@ -10,6 +10,10 @@ late, rows a reorg replaced (same id, new fingerprint), and rows a reorg dropped (missing from a fully fetched range). The checkpoint advances only past chunks that were fetched completely. + +Every stream reconciles against its own inbox and checkpoints. Streams in one +process pass a shared ``heads`` reader so ``chain_metadata`` is read once per +cycle, not once per stream. """ import time @@ -17,11 +21,12 @@ from dataclasses import dataclass, field from typing import Any -from protocols.timelock.subscriber.envio import ChainHead, EnvioQueryError, GraphQLPost, fetch_chain_heads, fetch_range -from protocols.timelock.subscriber.inbox import SOURCE_HTTP, EventInbox from utils.logger import get_logger +from utils.subscriber.envio import ChainHead, EnvioQueryError, GraphQLPost, fetch_chain_heads, fetch_range +from utils.subscriber.inbox import SOURCE_HTTP, EventInbox +from utils.subscriber.stream import EventStream -_logger = get_logger("timelock.subscriber.reconcile") +_logger = get_logger("subscriber.reconcile") # Blocks re-read below the checkpoint on every cycle, about 15-20 minutes per # chain: deeper than any reorg seen on these chains, and long enough to cover an @@ -29,6 +34,8 @@ REORG_OVERLAP_BLOCKS: dict[int, int] = { 1: 64, 137: 512, + 999: 1200, # HyperEVM, ~1s blocks + 4663: 10_000, # Robinhood Chain, ~0.1s blocks 8453: 600, 42161: 4800, 747474: 1200, @@ -41,6 +48,8 @@ MAX_BLOCK_SPAN: dict[int, int] = { 1: 50_000, 137: 300_000, + 999: 600_000, + 4663: 6_000_000, 8453: 300_000, 42161: 2_400_000, 747474: 600_000, @@ -101,21 +110,21 @@ def ok(self) -> bool: return self.error is None and all(c.error is None for c in self.chains) -def build_scope(timelocks: Iterable[Any]) -> dict[int, set[str]]: - """Map chain id to the lowercase timelock addresses monitored on it.""" +def build_scope(contracts: Iterable[Any]) -> dict[int, set[str]]: + """Map chain id to the lowercase addresses monitored on it (objects with ``address`` and ``chain_id``).""" scope: dict[int, set[str]] = {} - for timelock in timelocks: - scope.setdefault(int(timelock.chain_id), set()).add(str(timelock.address).lower()) + for contract in contracts: + scope.setdefault(int(contract.chain_id), set()).add(str(contract.address).lower()) return scope -def in_scope(event: dict[str, Any], scope: dict[int, set[str]]) -> bool: - """True when the row's (chain, timelock) pair is monitored. +def in_scope(event: dict[str, Any], scope: dict[int, set[str]], stream: EventStream) -> bool: + """True when the row's (chain, contract) pair is monitored. The same address is monitored on some chains only, so a query filtered by address and chain separately can return pairs outside the scope. """ - return str(event.get("timelockAddress", "")).lower() in scope.get(int(event.get("chainId", -1)), set()) + return str(event.get(stream.address_field, "")).lower() in scope.get(int(event.get("chainId", -1)), set()) def reconcile_chain( @@ -142,7 +151,15 @@ def reconcile_chain( if config.start_lookback_seconds > 0: since_ts = int(now()) - config.start_lookback_seconds rows, pages = fetch_range( - post, chain_id, addresses, -1, head_block, config.page_size, config.max_pages, since_ts=since_ts + post, + inbox.stream, + chain_id, + addresses, + -1, + head_block, + config.page_size, + config.max_pages, + since_ts=since_ts, ) _fold(inbox, rows, result, pages, now()) inbox.set_checkpoint(chain_id, head_block, head_block, now()) @@ -164,7 +181,9 @@ def reconcile_chain( lower = start for _ in range(config.max_chunks_per_cycle): upper = min(head_block, cursor + span) - rows, pages = fetch_range(post, chain_id, addresses, lower, upper, config.page_size, config.max_pages) + rows, pages = fetch_range( + post, inbox.stream, chain_id, addresses, lower, upper, config.page_size, config.max_pages + ) observed_at = now() seen = _fold(inbox, rows, result, pages, observed_at) removed = inbox.mark_missing(chain_id, lower + 1, upper, seen, SOURCE_HTTP, observed_at) @@ -201,23 +220,28 @@ def reconcile_all( scope: dict[int, set[str]], config: ReconcileConfig, now: Callable[[], float] = time.time, + heads: Callable[[], dict[int, ChainHead]] | None = None, ) -> ReconcileReport: - """Run one reconcile cycle over every monitored chain and log its timing. + """Run one reconcile cycle over every monitored chain of ``inbox``'s stream and log its timing. A failure on one chain is recorded and the others still run, so each chain's checkpoint advances on its own. + + Args: + heads: Chain-head reader shared with other streams (``SharedChainHeads``); + defaults to reading ``chain_metadata`` for this cycle alone. """ report = ReconcileReport(started_at=now()) clock = time.perf_counter() try: - heads = fetch_chain_heads(post) + chain_heads = heads() if heads is not None else fetch_chain_heads(post) except EnvioQueryError as exc: report.error = str(exc) - heads = {} + chain_heads = {} if report.error is None: for chain_id in sorted(scope): - head = heads.get(chain_id) + head = chain_heads.get(chain_id) if head is None: report.chains.append(ChainReconcileResult(chain_id=chain_id, error="no chain_metadata row")) continue diff --git a/protocols/timelock/subscriber/report.py b/utils/subscriber/report.py similarity index 96% rename from protocols/timelock/subscriber/report.py rename to utils/subscriber/report.py index 0916553d..fa4fbb90 100644 --- a/protocols/timelock/subscriber/report.py +++ b/utils/subscriber/report.py @@ -1,4 +1,4 @@ -"""Coverage and latency report for the timelock subscriber pilot. +"""Coverage and latency report for one stream of the event subscriber pilot. Answers the pilot's questions from the inbox alone: @@ -20,7 +20,7 @@ from dataclasses import asdict, dataclass, field from typing import Any -from protocols.timelock.subscriber.inbox import EventInbox +from utils.subscriber.inbox import EventInbox # Below this many events the latency percentiles mean little; keep collecting. MIN_EVENTS_FOR_DECISION = 5 @@ -59,6 +59,8 @@ class PilotReport: window_start: float window_end: float + # Noun of the stream the report covers ("timelock", "Morpho governance"). + stream: str = "" events: int = 0 confirmed_by_http: int = 0 pending_http: int = 0 @@ -178,7 +180,7 @@ def build_report(inbox: EventInbox, hours: float, now: float | None = None) -> P """Compute the pilot report over events with block timestamps in the last ``hours``.""" end = now if now is not None else time.time() start = end - hours * 3600 - report = PilotReport(window_start=start, window_end=end) + report = PilotReport(window_start=start, window_end=end, stream=inbox.stream.noun) checkpoints = {cp.chain_id: cp.block_number for cp in inbox.checkpoints()} ws_delays: list[float] = [] @@ -278,7 +280,7 @@ def lat(stats: LatencyStats, unit: str = "s") -> str: f"{time.strftime('%Y-%m-%d %H:%M', time.gmtime(report.window_end))} UTC" ) lines = [ - f"Timelock subscriber pilot report ({window})", + f"{report.stream[:1].upper()}{report.stream[1:]} subscriber pilot report ({window})", "", "Coverage", f" events: {report.events} (confirmed by HTTP {report.confirmed_by_http}, pending HTTP {report.pending_http})", diff --git a/utils/subscriber/stream.py b/utils/subscriber/stream.py new file mode 100644 index 00000000..5cc8fd24 --- /dev/null +++ b/utils/subscriber/stream.py @@ -0,0 +1,64 @@ +"""Named event streams the continuous subscriber can follow. + +A stream is one Envio entity of event occurrences (``TimelockEvent``, +``MorphoGovernanceEvent``, ...) plus where its inbox lives. Every stream shares +the transport, inbox, reconciliation, and report code, and keeps its own SQLite +file, checkpoints, and provenance, so one stream's backlog or failure never +holds back another. Streams are defined next to the protocol they serve. +""" + +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field +from typing import Any + + +@dataclass(frozen=True) +class StreamLabel: + """Human label for one monitored contract, shown in pilot notices.""" + + label: str + protocol: str + + +@dataclass(frozen=True) +class EventStream: + """How to query, store, and describe one Envio event entity. + + Every entity must expose ``id`` (``__``), + ``chainId``, ``blockNumber``, ``blockHash``, ``blockTimestamp``, ``logIndex``, + ``transactionHash``, and ``eventName`` besides its own fields. + """ + + # Short identifier for the CLI, logs, and the notice label ("-subscriber"). + name: str + # Noun used in notice and report text ("timelock", "Morpho governance"). + noun: str + # Envio GraphQL entity, e.g. "TimelockEvent". + entity: str + # Entity field holding the monitored contract address. + address_field: str + # Notice label for that contract ("Timelock", "Vault"). + address_label: str + # GraphQL selection set for the entity (both transports use it, so rows fingerprint identically). + fields: str + db_filename: str + # SQLite tables are ``_events``, ``_changes``, ... + table_prefix: str + # Optional entity field stored as the inbox ``operation_id`` and shown in notices. + operation_field: str | None = None + # Extra notice lines for one event row payload (e.g. decoded arguments). + describe: Callable[[dict[str, Any]], list[str]] | None = field(default=None, compare=False) + + def table(self, name: str) -> str: + """Return this stream's SQLite table called ``name`` (events, changes, checkpoints, log, status).""" + return f"{self.table_prefix}_{name}" + + +@dataclass(frozen=True) +class StreamScope: + """The contracts one stream monitors, and how notices name them.""" + + # chain id -> lowercase monitored addresses + addresses: dict[int, set[str]] + # (lowercase address, chain id) -> label + labels: Mapping[tuple[str, int], StreamLabel] diff --git a/protocols/timelock/subscriber/worker.py b/utils/subscriber/worker.py similarity index 70% rename from protocols/timelock/subscriber/worker.py rename to utils/subscriber/worker.py index e471f680..49a8cb5a 100644 --- a/protocols/timelock/subscriber/worker.py +++ b/utils/subscriber/worker.py @@ -1,6 +1,6 @@ -"""Continuous timelock event worker: WebSocket subscription, HTTP reconciliation, heartbeat. +"""Continuous event worker: WebSocket subscription, HTTP reconciliation, heartbeat. -Three asyncio tasks share one ``EventInbox``: +One ``Worker`` follows one event stream. Three asyncio tasks share its ``EventInbox``: - ``ws``: holds a ``graphql-transport-ws`` live query over the newest monitored events and folds every snapshot into the inbox, reconnecting with capped @@ -11,6 +11,12 @@ advancing, and pings the systemd watchdog only while reconciliation keeps completing, so a wedged worker is restarted. +``run_workers`` runs several streams in one process: one set of signal handlers, +one systemd ``READY``, and a watchdog ping only while every stream is alive, so a +wedged stream restarts the process instead of hiding behind a healthy one. The +streams share the HTTP poster and a ``SharedChainHeads`` reader, and each keeps +its own WebSocket connection, inbox, and checkpoints. + 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. @@ -24,15 +30,15 @@ from dataclasses import dataclass from typing import Any -from protocols.timelock.subscriber.envio import GraphQLPost, address_variants, live_subscription -from protocols.timelock.subscriber.inbox import SOURCE_HTTP, SOURCE_WS, EventInbox -from protocols.timelock.subscriber.notices import PilotNotices -from protocols.timelock.subscriber.reconcile import ReconcileConfig, ReconcileReport, in_scope, reconcile_all 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.inbox import SOURCE_HTTP, SOURCE_WS, EventInbox +from utils.subscriber.notices import PilotNotices +from utils.subscriber.reconcile import ReconcileConfig, ReconcileReport, in_scope, reconcile_all -_logger = get_logger("timelock.subscriber.worker") +_logger = get_logger("subscriber.worker") TRANSPORTS = ("both", "ws", "http") @@ -64,7 +70,7 @@ class WorkerConfig: class Worker: - """Runs the subscriber loops until ``stop()`` (or SIGTERM/SIGINT).""" + """Runs one stream's subscriber loops until ``stop()`` (or SIGTERM/SIGINT).""" def __init__( self, @@ -76,6 +82,7 @@ def __init__( subscribe_fn: SubscribeFn = subscribe, now: Callable[[], float] = time.time, notices: PilotNotices | None = None, + heads: Callable[[], dict[int, ChainHead]] | None = None, ) -> None: if config.transport not in TRANSPORTS: raise ValueError(f"transport must be one of {TRANSPORTS}, got {config.transport!r}") @@ -89,6 +96,10 @@ def __init__( self.subscribe_fn = subscribe_fn self.now = now self.notices = notices + self.heads = heads + self.stream = inbox.stream + # False when ``run_workers`` owns signals and systemd notifications for the process. + self.supervised = True self._stop = asyncio.Event() self.ws_connected = False self.ws_connects = 0 @@ -113,12 +124,8 @@ async def _sleep(self, seconds: float) -> None: async def run(self) -> None: """Run the configured loops until stopped.""" - loop = asyncio.get_running_loop() - for sig in (signal.SIGTERM, signal.SIGINT): - try: - loop.add_signal_handler(sig, self.stop) - except NotImplementedError, RuntimeError: - pass + if self.supervised: + _install_signal_handlers(self.stop) self.started_at = self.now() await asyncio.to_thread(self.inbox.log, "worker_started", self.started_at, transport=self.config.transport) if self.notices is not None: @@ -136,11 +143,13 @@ async def run(self) -> None: helpers.append(asyncio.create_task(self.notices.notifier.run(), name="notifier")) if self.config.transport in ("both", "ws"): self.notices.ws_disconnected(self.started_at, "not connected yet") - notify("READY=1", f"STATUS=transport={self.config.transport}") + if self.supervised: + notify("READY=1", f"STATUS=transport={self.config.transport}") try: await self._stop.wait() finally: - notify("STOPPING=1") + if self.supervised: + notify("STOPPING=1") for task in [*loops, *helpers]: task.cancel() await asyncio.gather(*loops, *helpers, return_exceptions=True) @@ -149,7 +158,7 @@ async def run(self) -> None: async def ws_loop(self) -> None: """Hold the live subscription, reconnecting with capped exponential backoff.""" addresses = address_variants({a for addrs in self.scope.values() for a in addrs}) - query = live_subscription(self.config.snapshot_limit, sorted(self.scope)) + query = live_subscription(self.stream, self.config.snapshot_limit, sorted(self.scope)) failures = 0 while not self._stop.is_set(): connected_at: float | None = None @@ -163,7 +172,7 @@ async def ws_loop(self) -> None: self.ws_connected = True self.ws_connects += 1 await asyncio.to_thread(self.inbox.log, "ws_connected", received_at) - _logger.info("WebSocket subscription live") + _logger.info("%s WebSocket subscription live", self.stream.name) if self.notices is not None: self.notices.ws_connected(received_at) if payload.get("errors"): @@ -189,12 +198,14 @@ async def ws_loop(self) -> None: failures += 1 delay = backoff_delay(failures, self.config.backoff_initial, self.config.backoff_max) delay *= random.uniform(0.8, 1.2) - _logger.warning("WebSocket disconnected (%s); reconnecting in %.1fs", error, delay) + _logger.warning("%s WebSocket disconnected (%s); reconnecting in %.1fs", self.stream.name, error, delay) 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("TimelockEvent") or [] if in_scope(r, self.scope)] + rows = [ + r for r in (payload.get("data") or {}).get(self.stream.entity) or [] 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 @@ -208,10 +219,11 @@ async def _ingest_snapshot(self, payload: dict[str, Any], received_at: float) -> for event_id in result.new_ids: row = by_id[event_id] _logger.info( - "ws new event %s chain=%s timelock=%s %s latency=%.1fs", + "%s ws new event %s chain=%s contract=%s %s latency=%.1fs", + self.stream.name, event_id, row.get("chainId"), - row.get("timelockAddress"), + row.get(self.stream.address_field), row.get("eventName"), received_at - int(row["blockTimestamp"]), ) @@ -233,12 +245,12 @@ async def reconcile_loop(self) -> None: ws_was_connected = self.ws_connected try: report = await asyncio.to_thread( - reconcile_all, self.post, self.inbox, self.scope, self.config.reconcile, self.now + reconcile_all, self.post, self.inbox, self.scope, self.config.reconcile, self.now, self.heads ) except asyncio.CancelledError: raise except Exception as exc: # noqa: BLE001 - an unexpected bug must not end reconciliation - _logger.exception("reconcile cycle crashed") + _logger.exception("%s reconcile cycle crashed", self.stream.name) report = ReconcileReport(started_at=started, error=f"{type(exc).__name__}: {exc}") self.last_reconcile = report self.last_reconcile_at = self.now() @@ -249,19 +261,21 @@ async def reconcile_loop(self) -> None: await self._sleep(self.config.reconcile_interval - (self.now() - started)) def _log_reconcile(self, report: ReconcileReport, ws_expected: bool) -> None: + """Log one reconcile cycle's errors, WebSocket misses, and new events.""" + name = self.stream.name if report.error: - _logger.error("reconcile failed: %s", report.error) + _logger.error("%s reconcile failed: %s", name, report.error) for chain in report.chains: if chain.error: - _logger.error("reconcile chain %s failed: %s", chain.chain_id, chain.error) + _logger.error("%s reconcile chain %s failed: %s", name, chain.chain_id, chain.error) if chain.new and ws_expected: # HTTP found rows first although the subscription was up: a WS miss. _logger.warning( - "chain %s: HTTP found %s events before WS: %s", chain.chain_id, chain.new, chain.new_ids + "%s chain %s: HTTP found %s events before WS: %s", name, chain.chain_id, chain.new, chain.new_ids ) elif chain.new: - _logger.info("chain %s: %s new events via %s", chain.chain_id, chain.new, SOURCE_HTTP) - _logger.info("reconcile done in %.0fms (ok=%s)", report.duration_ms, report.ok) + _logger.info("%s chain %s: %s new events via %s", name, chain.chain_id, chain.new, SOURCE_HTTP) + _logger.info("%s reconcile done in %.0fms (ok=%s)", name, report.duration_ms, report.ok) def health(self) -> dict[str, Any]: """Return the worker's current health snapshot (also written as the heartbeat).""" @@ -283,6 +297,7 @@ def health(self) -> dict[str, Any]: last = self.last_reconcile return { "at": now, + "stream": self.stream.name, "transport": self.config.transport, "ws_connected": self.ws_connected, "ws_connects": self.ws_connects, @@ -321,7 +336,8 @@ async def heartbeat_loop(self) -> None: chain_id = chain["chain_id"] if chain["stalled"] and chain_id not in self._stalled: _logger.warning( - "chain %s indexed head %s has not advanced for %.0fs", + "%s chain %s indexed head %s has not advanced for %.0fs", + self.stream.name, chain_id, chain["indexed_head"], chain["head_age_seconds"], @@ -330,7 +346,7 @@ async def heartbeat_loop(self) -> None: if self.notices is not None: self.notices.stall_changed(chain_id, True, chain["indexed_head"], chain["head_age_seconds"]) elif not chain["stalled"] and chain_id in self._stalled: - _logger.info("chain %s indexed head advancing again", chain_id) + _logger.info("%s chain %s indexed head advancing again", self.stream.name, chain_id) self._stalled.discard(chain_id) if self.notices is not None: self.notices.stall_changed(chain_id, False, chain["indexed_head"], 0.0) @@ -340,12 +356,67 @@ async def heartbeat_loop(self) -> None: await self.notices.maybe_daily_summary(health["at"]) except Exception: # noqa: BLE001 - a report bug must not stop the heartbeat _logger.warning("daily summary failed", exc_info=True) - if self.is_alive(): - notify( - "WATCHDOG=1", f"STATUS=ws={'up' if self.ws_connected else 'down'} stalled={sorted(self._stalled)}" - ) - else: + # Under run_workers the supervisor pings the watchdog for the whole process. + if self.supervised and self.is_alive(): + notify("WATCHDOG=1", f"STATUS={self.status_line()}") + elif self.supervised: _logger.error( "a worker loop stopped or reconcile has not completed recently; withholding watchdog ping" ) await self._sleep(self.config.heartbeat_interval) + + def status_line(self) -> str: + """Short systemd ``STATUS`` text for this stream.""" + return f"ws={'up' if self.ws_connected else 'down'} stalled={sorted(self._stalled)}" + + +def _install_signal_handlers(stop: Callable[[], None]) -> None: + """Call ``stop`` on SIGTERM/SIGINT where the event loop supports it.""" + loop = asyncio.get_running_loop() + for sig in (signal.SIGTERM, signal.SIGINT): + try: + loop.add_signal_handler(sig, stop) + except NotImplementedError, RuntimeError: + pass + + +async def run_workers(workers: list[Worker], watchdog_interval: float = 30.0) -> None: + """Run several stream workers in one process until any of them stops or a signal arrives. + + Signals stop every worker. systemd gets one ``READY`` and a ``WATCHDOG`` ping only + while every worker is alive, so one wedged stream restarts the whole unit. + """ + if not workers: + raise ValueError("run_workers needs at least one worker") + for worker in workers: + worker.supervised = False + + def stop_all() -> None: + for worker in workers: + worker.stop() + + _install_signal_handlers(stop_all) + runs = [asyncio.create_task(worker.run(), name=f"worker-{worker.stream.name}") for worker in workers] + notify("READY=1", "STATUS=" + " ".join(f"{w.stream.name}:transport={w.config.transport}" for w in workers)) + + async def watchdog() -> None: + while True: + if all(worker.is_alive() for worker in workers): + notify("WATCHDOG=1", "STATUS=" + " ".join(f"{w.stream.name}:{w.status_line()}" for w in workers)) + else: + dead = [w.stream.name for w in workers if not w.is_alive()] + _logger.error("streams %s stopped or stopped reconciling; withholding watchdog ping", dead) + await asyncio.sleep(watchdog_interval) + + pinger = asyncio.create_task(watchdog(), name="watchdog") + try: + # A worker only returns once stopped (or crashed); stop the rest with it. + await asyncio.wait(runs, return_when=asyncio.FIRST_COMPLETED) + finally: + notify("STOPPING=1") + stop_all() + pinger.cancel() + results = await asyncio.gather(*runs, pinger, return_exceptions=True) + for result in results[: len(runs)]: + if isinstance(result, Exception): + raise result # surface a crashed stream to run_with_alert From aead53386aeee81e87c3e4a63e0598920545f67b Mon Sep 17 00:00:00 2001 From: spalen0 Date: Mon, 26 Oct 2026 17:35:25 +0100 Subject: [PATCH 2/2] Fix watchdog recovery after subscriber heartbeat failures --- tests/test_event_subscriber_streams.py | 130 +++++++++++++++++++++++++ utils/subscriber/worker.py | 25 +++-- 2 files changed, 146 insertions(+), 9 deletions(-) diff --git a/tests/test_event_subscriber_streams.py b/tests/test_event_subscriber_streams.py index cb2cfa85..3143d3ef 100644 --- a/tests/test_event_subscriber_streams.py +++ b/tests/test_event_subscriber_streams.py @@ -319,6 +319,136 @@ async def crash() -> None: asyncio.run(scenario()) +def test_run_workers_withholds_watchdog_after_heartbeat_write_fails(tmp_path: Path) -> None: + """A SQLite heartbeat failure must stop pings while reconciliation continues.""" + notified: list[tuple[str, ...]] = [] + + async def scenario() -> None: + envio = FakeEnvio() + envio.set_head(1, 1) + inbox, _ = _inboxes(tmp_path) + config = WorkerConfig(transport="http", reconcile_interval=0.05, heartbeat_interval=0.02, reconcile=CONFIG) + worker = Worker(envio.post, None, inbox, TIMELOCK_SCOPE, config) + task = asyncio.create_task(run_workers([worker], watchdog_interval=0.01)) + try: + await _until(lambda: "heartbeat" in inbox.status() and any(m[0] == "WATCHDOG=1" for m in notified)) + with patch.object(inbox, "set_status", side_effect=sqlite3.OperationalError("database is locked")) as write: + await _until(lambda: write.called) + await _until(lambda: not worker.is_alive()) + last_reconcile = worker.last_reconcile_at + notified.clear() + await _until(lambda: worker.last_reconcile_at != last_reconcile) + await asyncio.sleep(0.05) + assert not any(m[0] == "WATCHDOG=1" for m in notified) + finally: + worker.stop() + await asyncio.wait_for(task, 5) + + with patch("utils.subscriber.worker.notify", side_effect=lambda *a: notified.append(a)): + asyncio.run(scenario()) + + +@pytest.mark.parametrize("transport", ["http", "ws"]) +@pytest.mark.parametrize("after_first", [False, True], ids=["first-cycle", "later-cycle"]) +def test_run_workers_withholds_watchdog_when_heartbeat_hangs(tmp_path: Path, transport: str, after_first: bool) -> None: + """A pending heartbeat gets a bounded grace period, even with a healthy transport.""" + notified: list[tuple[str, ...]] = [] + blocked = threading.Event() + release = threading.Event() + + async def scenario() -> None: + envio = FakeEnvio() + envio.set_head(1, 1) + server = FakeWSServer(envio) + await server.start() + inbox, _ = _inboxes(tmp_path) + config = WorkerConfig(transport=transport, reconcile_interval=0.05, heartbeat_interval=0.02, reconcile=CONFIG) + worker = Worker(envio.post, server.url, inbox, TIMELOCK_SCOPE, config) + set_status = inbox.set_status + writes = 0 + + def slow_write(key: str, value: dict, at: float) -> None: + nonlocal writes + writes += 1 + if after_first and writes == 1: + clock.return_value = 1_040.0 + if writes > int(after_first): + blocked.set() + if not release.wait(5): + raise TimeoutError("test did not release the blocked heartbeat write") + set_status(key, value, at) + + with patch.object(inbox, "set_status", side_effect=slow_write): + task = asyncio.create_task(run_workers([worker], watchdog_interval=0.01)) + try: + await _until(blocked.is_set) + await _until(lambda: any(m[0] == "WATCHDOG=1" for m in notified)) + deadline = 3 * config.heartbeat_interval + 120 + last_progress = 1_040 if after_first else 1_000 + clock.return_value = last_progress + deadline - 1 + assert worker.is_alive() + clock.return_value = last_progress + deadline + assert not worker.is_alive() + with patch.object(worker, "now", return_value=worker.now() - 3_600): + assert not worker.is_alive() + notified.clear() + last_reconcile = worker.last_reconcile_at + if transport == "http": + await _until(lambda: worker.last_reconcile_at != last_reconcile) + await asyncio.sleep(0.05) + assert not any(m[0] == "WATCHDOG=1" for m in notified) + release.set() + await _until(lambda: any(m[0] == "WATCHDOG=1" for m in notified)) + assert worker.is_alive() + finally: + release.set() + worker.stop() + await asyncio.wait_for(task, 5) + await server.stop() + + with ( + patch("utils.subscriber.worker.notify", side_effect=lambda *a: notified.append(a)), + patch("utils.subscriber.worker.monotonic", return_value=1_000.0) as clock, + ): + asyncio.run(scenario()) + + +def test_run_workers_keeps_pinging_after_notifier_crashes(tmp_path: Path) -> None: + """Notifier failure must not prevent fresh heartbeats from feeding the watchdog.""" + notified: list[tuple[str, ...]] = [] + + async def scenario() -> None: + envio = FakeEnvio() + envio.set_head(1, 1) + inbox, _ = _inboxes(tmp_path) + notifier = RecordingNotifier() + notices = PilotNotices(notifier, inbox, {}) + config = WorkerConfig(transport="http", reconcile_interval=0.05, heartbeat_interval=0.02, reconcile=CONFIG) + worker = Worker(envio.post, None, inbox, TIMELOCK_SCOPE, config, notices=notices) + crashed = asyncio.Event() + + async def crash() -> None: + crashed.set() + raise RuntimeError("notifier failed") + + with patch.object(notifier, "run", side_effect=crash): + task = asyncio.create_task(run_workers([worker], watchdog_interval=0.01)) + try: + await asyncio.wait_for(crashed.wait(), 5) + await _until(lambda: "heartbeat" in inbox.status()) + notified.clear() + last_heartbeat = inbox.status()["heartbeat"][1] + await _until(lambda: inbox.status()["heartbeat"][1] != last_heartbeat) + await _until(lambda: any(m[0] == "WATCHDOG=1" for m in notified)) + assert worker.is_alive() + finally: + worker.stop() + await asyncio.wait_for(task, 5) + + with patch("utils.subscriber.worker.notify", side_effect=lambda *a: notified.append(a)): + 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) diff --git a/utils/subscriber/worker.py b/utils/subscriber/worker.py index 49a8cb5a..b7ec4e68 100644 --- a/utils/subscriber/worker.py +++ b/utils/subscriber/worker.py @@ -28,6 +28,7 @@ import time from collections.abc import AsyncIterator, Callable from dataclasses import dataclass +from time import monotonic from typing import Any from utils.graphql_ws import GraphQLWSError, subscribe @@ -109,6 +110,7 @@ def __init__( self.last_reconcile: ReconcileReport | None = None self._stalled: set[int] = set() self.started_at: float | None = None + self._heartbeat_progress_monotonic: float | None = None self._tasks: list[asyncio.Task[None]] = [] def stop(self) -> None: @@ -127,6 +129,7 @@ async def run(self) -> None: if self.supervised: _install_signal_handlers(self.stop) self.started_at = self.now() + self._heartbeat_progress_monotonic = monotonic() await asyncio.to_thread(self.inbox.log, "worker_started", self.started_at, transport=self.config.transport) if self.notices is not None: # Before any loop runs, so the first WebSocket snapshot is judged against it. @@ -136,8 +139,9 @@ async def run(self) -> None: loops.append(asyncio.create_task(self.reconcile_loop(), name="reconcile")) 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")) self._tasks = loops - helpers = [asyncio.create_task(self.heartbeat_loop(), name="heartbeat")] + helpers: list[asyncio.Task[None]] = [] if self.notices is not None: # Not in self._tasks: a dead notifier must not trip the watchdog. helpers.append(asyncio.create_task(self.notices.notifier.run(), name="notifier")) @@ -312,14 +316,18 @@ def health(self) -> dict[str, Any]: } def is_alive(self) -> bool: - """True while every loop task runs and reconcile cycles keep completing. + """True while required tasks run and heartbeat/reconcile cycles keep completing. - A loop task that ended (it should only end on stop) makes the worker dead. The - reconcile deadline counts from the worker's start until the first cycle - completes, so a first cycle that hangs is caught too. + A loop task that ended (it should only end on stop) makes the worker dead. + Deadlines count from the worker's start until the first cycle completes, + so a first cycle that hangs is caught too. Heartbeat progress uses a + monotonic clock so wall-clock adjustments cannot hide a stalled loop. """ if any(task.done() for task in self._tasks): return False + heartbeat = self._heartbeat_progress_monotonic + if heartbeat is not None and monotonic() >= heartbeat + 3 * self.config.heartbeat_interval + 120: + return False if self.config.transport == "ws": return True last = self.last_reconcile_at if self.last_reconcile_at is not None else self.started_at @@ -356,13 +364,12 @@ async def heartbeat_loop(self) -> None: await self.notices.maybe_daily_summary(health["at"]) except Exception: # noqa: BLE001 - a report bug must not stop the heartbeat _logger.warning("daily summary failed", exc_info=True) + self._heartbeat_progress_monotonic = monotonic() # Under run_workers the supervisor pings the watchdog for the whole process. if self.supervised and self.is_alive(): notify("WATCHDOG=1", f"STATUS={self.status_line()}") elif self.supervised: - _logger.error( - "a worker loop stopped or reconcile has not completed recently; withholding watchdog ping" - ) + _logger.error("a worker loop stopped or heartbeat/reconcile is stale; withholding watchdog ping") await self._sleep(self.config.heartbeat_interval) def status_line(self) -> str: @@ -405,7 +412,7 @@ async def watchdog() -> None: notify("WATCHDOG=1", "STATUS=" + " ".join(f"{w.stream.name}:{w.status_line()}" for w in workers)) else: dead = [w.stream.name for w in workers if not w.is_alive()] - _logger.error("streams %s stopped or stopped reconciling; withholding watchdog ping", dead) + _logger.error("streams %s stopped or heartbeat/reconcile is stale; withholding watchdog ping", dead) await asyncio.sleep(watchdog_interval) pinger = asyncio.create_task(watchdog(), name="watchdog")