diff --git a/.env.example b/.env.example index 830c4460..e97257d1 100644 --- a/.env.example +++ b/.env.example @@ -133,3 +133,11 @@ LOG_LEVEL=INFO # DEBUG, INFO, WARNING, ERROR (DEBUG skips Telegram sends) REQUEST_TIMEOUT=30 RETRY_COUNT=3 BACKOFF_FACTOR=1.0 + +# Independent emergency destinations. Unset secret disables only that destination. +# LIQUIDITY_WEBHOOK_URL=http://127.0.0.1:8080/webhook/emergency +# LIQUIDITY_WEBHOOK_SECRET=separate-liquidity-shared-secret +# VAULTS_MOMMY_WEBHOOK_URL=http://127.0.0.1:8081/webhook/emergency +# VAULTS_MOMMY_WEBHOOK_SECRET=separate-vaults-mommy-shared-secret +# Durable outbox defaults to $CACHE_DIR/emergency_dispatch.sqlite3. +# EMERGENCY_DISPATCH_DB=/srv/cache/emergency_dispatch.sqlite3 diff --git a/.gitignore b/.gitignore index 064cbecf..484a102b 100644 --- a/.gitignore +++ b/.gitignore @@ -55,3 +55,6 @@ label-cache/ monitoring.db monitoring.db-wal monitoring.db-shm + +# Durable emergency webhook outbox (including SQLite sidecars) +emergency_dispatch.sqlite3* diff --git a/README.md b/README.md index 6646fa88..73af15b7 100644 --- a/README.md +++ b/README.md @@ -109,3 +109,9 @@ For more details about this repository, check out AI generated docs using DeepWi ## License This repository is licensed under the GNU Affero General Public License v3.0. See [LICENSE](./LICENSE). + +## Emergency webhook delivery + +[Durable emergency delivery](utils/dispatch.md) documents the independent liquidity-monitoring +and optional vaults-mommy destinations, signed payload compatibility, severity-aware cooldowns, +recurring retries, persistent outbox and explicit recovery of terminal errors. diff --git a/automation/jobs.yaml b/automation/jobs.yaml index e085a2bc..e4fdced8 100644 --- a/automation/jobs.yaml +++ b/automation/jobs.yaml @@ -18,6 +18,12 @@ profiles: + emergency_delivery: + cron: "* * * * *" + description: "Drain durable emergency webhook retries even when no new alert fires." + tasks: + - { name: "emergency-delivery-retries", script: utils/dispatch_retry.py } + half_hourly: cron: "*/30 * * * *" description: "Thirty-minute risk monitoring for active Yearn strategies, and timelock alerts." diff --git a/tests/conftest.py b/tests/conftest.py index 66eb5125..35e6fe62 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -37,7 +37,7 @@ def _isolate_from_live_apis(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> """ for key in list(os.environ): if key in {"ETHERSCAN_TOKEN", "LIQUIDITY_WEBHOOK_SECRET", "TENDERLY_API_KEY"} or key.startswith( - ("PROVIDER_URL_", "TELEGRAM_", "LIQUIDITY_WEBHOOK_") + ("PROVIDER_URL_", "TELEGRAM_", "LIQUIDITY_WEBHOOK_", "VAULTS_MOMMY_WEBHOOK_", "EMERGENCY_DISPATCH_") ): monkeypatch.delenv(key, raising=False) monkeypatch.setenv("LOG_LEVEL", "INFO") diff --git a/tests/test_dispatch.py b/tests/test_dispatch.py index 80546281..595e0e2b 100644 --- a/tests/test_dispatch.py +++ b/tests/test_dispatch.py @@ -1,236 +1,304 @@ -"""Tests for dispatch.""" +"""Durable sender behavior: fanout, escalation, retry and legacy liquidity compatibility.""" import hashlib import hmac import json -import os -import unittest -from unittest.mock import MagicMock, patch +import multiprocessing +import sqlite3 +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock +import pytest import requests +from automation.config import load_jobs_config +from utils import dispatch from utils.alert import Alert, AlertSeverity - - -class TestDispatch(unittest.TestCase): - """Tests for the emergency dispatch utility.""" - - @patch("utils.dispatch.requests.post") - @patch("utils.dispatch._record_dispatch") - @patch("utils.dispatch._is_on_cooldown", return_value=False) - def test_dispatch_sends_for_3jane(self, mock_cooldown, mock_record, mock_post): - from utils.dispatch import dispatch_emergency_withdrawal - - mock_response = MagicMock() - mock_response.raise_for_status = MagicMock() - mock_post.return_value = mock_response - alert = Alert(severity=AlertSeverity.HIGH, message="Junior buffer low", protocol="3jane") - - with patch.dict(os.environ, {"LIQUIDITY_WEBHOOK_SECRET": "test_secret", "LOG_LEVEL": "INFO"}): - dispatch_emergency_withdrawal(alert) - - payload = json.loads(mock_post.call_args[1]["data"].decode("utf-8")) - self.assertEqual(payload["client_payload"]["protocol"], "3jane") - self.assertEqual(payload["client_payload"]["severity"], "HIGH") - mock_record.assert_called_once_with("3jane") - - @patch("utils.dispatch.requests.post") - @patch("utils.dispatch._record_dispatch") - @patch("utils.dispatch._is_on_cooldown", return_value=False) - def test_dispatch_sends_correct_payload(self, mock_cooldown, mock_record, mock_post): - from utils.dispatch import dispatch_emergency_withdrawal - - mock_response = MagicMock() - mock_response.raise_for_status = MagicMock() - mock_post.return_value = mock_response - - alert = Alert(severity=AlertSeverity.HIGH, message="Reserves low", protocol="infinifi") - - with patch.dict(os.environ, {"LIQUIDITY_WEBHOOK_SECRET": "test_secret", "LOG_LEVEL": "INFO"}): - dispatch_emergency_withdrawal(alert) - - mock_post.assert_called_once() - call_args = mock_post.call_args[0] - call_kwargs = mock_post.call_args[1] - self.assertEqual(call_args[0], "http://127.0.0.1:8080/webhook/emergency") - body = call_kwargs["data"] - self.assertIsInstance(body, bytes) - self.assertNotIn("json", call_kwargs) - payload = json.loads(body.decode("utf-8")) - self.assertEqual(payload["event_type"], "emergency_withdrawal") - self.assertEqual(payload["client_payload"]["protocol"], "infinifi") - self.assertEqual(payload["client_payload"]["severity"], "HIGH") - self.assertEqual(payload["client_payload"]["message"], "Reserves low") - # Payload should only contain protocol, severity, and message (no markets/vault/chain) - self.assertEqual(set(payload["client_payload"].keys()), {"protocol", "severity", "message"}) - - headers = call_kwargs["headers"] - expected_hmac = hmac.new(b"test_secret", body, hashlib.sha256).hexdigest() - self.assertEqual(headers["X-Hub-Signature-256"], f"sha256={expected_hmac}") - self.assertEqual(headers["Content-Type"], "application/json") - - mock_record.assert_called_once_with("infinifi") - - @patch("utils.dispatch.requests.post") - @patch("utils.dispatch._record_dispatch") - @patch("utils.dispatch._is_on_cooldown", return_value=False) - def test_dispatch_uses_configured_webhook_url(self, mock_cooldown, mock_record, mock_post): - from utils.dispatch import dispatch_emergency_withdrawal - - mock_response = MagicMock() - mock_response.raise_for_status = MagicMock() - mock_post.return_value = mock_response - - alert = Alert(severity=AlertSeverity.HIGH, message="Reserves low", protocol="infinifi") - - with patch.dict( - os.environ, - { - "LIQUIDITY_WEBHOOK_SECRET": "test_secret", - "LIQUIDITY_WEBHOOK_URL": "http://localhost:9000/webhook/emergency", - "LOG_LEVEL": "INFO", - }, - ): - dispatch_emergency_withdrawal(alert) - - self.assertEqual(mock_post.call_args[0][0], "http://localhost:9000/webhook/emergency") - - @patch("utils.dispatch.requests.post") - def test_dispatch_skips_low_and_medium_severity(self, mock_post): - from utils.dispatch import dispatch_emergency_withdrawal - - for severity in (AlertSeverity.LOW, AlertSeverity.MEDIUM): - dispatch_emergency_withdrawal(Alert(severity=severity, message="m", protocol="infinifi")) - mock_post.assert_not_called() - - @patch("utils.dispatch.requests.post") - @patch("utils.dispatch._is_on_cooldown", return_value=False) - def test_dispatch_skips_unknown_protocol(self, mock_cooldown, mock_post): - from utils.dispatch import dispatch_emergency_withdrawal - - alert = Alert(severity=AlertSeverity.HIGH, message="alert", protocol="unknown_protocol") - - with patch.dict(os.environ, {"LIQUIDITY_WEBHOOK_SECRET": "test_secret"}): - dispatch_emergency_withdrawal(alert) - - mock_post.assert_not_called() - - @patch("utils.dispatch.requests.post") - @patch("utils.dispatch._is_on_cooldown", return_value=True) - def test_dispatch_skips_on_cooldown(self, mock_cooldown, mock_post): - from utils.dispatch import dispatch_emergency_withdrawal - - alert = Alert(severity=AlertSeverity.HIGH, message="alert", protocol="infinifi") - - with patch.dict(os.environ, {"LIQUIDITY_WEBHOOK_SECRET": "test_secret"}): - dispatch_emergency_withdrawal(alert) - - mock_post.assert_not_called() - - @patch("utils.dispatch.requests.post") - @patch("utils.dispatch._is_on_cooldown", return_value=False) - def test_dispatch_skips_missing_webhook_secret(self, mock_cooldown, mock_post): - from utils.dispatch import dispatch_emergency_withdrawal - - alert = Alert(severity=AlertSeverity.HIGH, message="alert", protocol="infinifi") - - with patch.dict(os.environ, {}, clear=True): - dispatch_emergency_withdrawal(alert) - - mock_post.assert_not_called() - - @patch("utils.dispatch.requests.post") - @patch("utils.dispatch._record_dispatch") - @patch("utils.dispatch._is_on_cooldown", return_value=False) - def test_dispatch_critical_sends_critical_severity(self, mock_cooldown, mock_record, mock_post): - from utils.dispatch import dispatch_emergency_withdrawal - - mock_response = MagicMock() - mock_response.raise_for_status = MagicMock() - mock_post.return_value = mock_response - - alert = Alert(severity=AlertSeverity.CRITICAL, message="total failure", protocol="infinifi") - - with patch.dict(os.environ, {"LIQUIDITY_WEBHOOK_SECRET": "test_secret", "LOG_LEVEL": "INFO"}): - dispatch_emergency_withdrawal(alert) - - payload = json.loads(mock_post.call_args[1]["data"].decode("utf-8")) - self.assertEqual(payload["client_payload"]["severity"], "CRITICAL") - - @patch("utils.dispatch.requests.post") - @patch("utils.dispatch._record_dispatch") - @patch("utils.dispatch._is_on_cooldown", return_value=False) - def test_dispatch_handles_request_exception(self, mock_cooldown, mock_record, mock_post): - from utils.dispatch import dispatch_emergency_withdrawal - - mock_post.side_effect = requests.RequestException("Connection error") - - alert = Alert(severity=AlertSeverity.HIGH, message="alert", protocol="infinifi") - - with patch.dict(os.environ, {"LIQUIDITY_WEBHOOK_SECRET": "test_secret", "LOG_LEVEL": "INFO"}): - # Should not raise - dispatch_emergency_withdrawal(alert) - - mock_record.assert_not_called() - - @patch("utils.dispatch.requests.post") - @patch("utils.dispatch._record_dispatch") - @patch("utils.dispatch._is_on_cooldown", return_value=False) - def test_dispatch_uses_protocol_not_channel(self, mock_cooldown, mock_record, mock_post): - """Dispatch uses alert.protocol (not channel) for payload and cooldown.""" - from utils.dispatch import dispatch_emergency_withdrawal - - mock_response = MagicMock() - mock_response.raise_for_status = MagicMock() - mock_post.return_value = mock_response - - alert = Alert(severity=AlertSeverity.HIGH, message="redeem value dropped", protocol="origin", channel="pegs") - - with patch.dict(os.environ, {"LIQUIDITY_WEBHOOK_SECRET": "test_secret", "LOG_LEVEL": "INFO"}): - dispatch_emergency_withdrawal(alert) - - payload = json.loads(mock_post.call_args[1]["data"].decode("utf-8")) - self.assertEqual(payload["client_payload"]["protocol"], "origin") - mock_record.assert_called_once_with("origin") - - @patch("utils.dispatch.requests.post") - @patch("utils.dispatch._is_on_cooldown", return_value=False) - def test_dispatch_skips_non_dispatchable_channel_protocol(self, mock_cooldown, mock_post): - """Protocol not in DISPATCHABLE_PROTOCOLS is skipped even with a valid channel.""" - from utils.dispatch import dispatch_emergency_withdrawal - - alert = Alert(severity=AlertSeverity.HIGH, message="peg alert", protocol="puffer", channel="pegs") - - with patch.dict(os.environ, {"LIQUIDITY_WEBHOOK_SECRET": "test_secret"}): - dispatch_emergency_withdrawal(alert) - - mock_post.assert_not_called() - - @patch("utils.dispatch.requests.post") - def test_dispatch_skips_in_debug_mode(self, mock_post): - from utils.dispatch import dispatch_emergency_withdrawal - - alert = Alert(severity=AlertSeverity.HIGH, message="alert", protocol="infinifi") - - with patch.dict(os.environ, {"LIQUIDITY_WEBHOOK_SECRET": "test_secret", "LOG_LEVEL": "DEBUG"}): - dispatch_emergency_withdrawal(alert) - - mock_post.assert_not_called() - - def test_cooldown_logic(self): - import time - - from utils.dispatch import _is_on_cooldown - - with patch("utils.dispatch.get_last_value_for_key_from_file") as mock_get: - # No previous dispatch - mock_get.return_value = 0 - self.assertFalse(_is_on_cooldown("infinifi")) - - # Recent dispatch (within cooldown) - mock_get.return_value = str(time.time() - 10) - self.assertTrue(_is_on_cooldown("infinifi", cooldown_seconds=60)) - - # Old dispatch (past cooldown) - mock_get.return_value = str(time.time() - 7200) - self.assertFalse(_is_on_cooldown("infinifi", cooldown_seconds=3600)) +from utils.dispatch_queue import Destination, DispatchQueue + + +@pytest.fixture +def configured(monkeypatch: pytest.MonkeyPatch) -> list[float]: + monkeypatch.setenv("LIQUIDITY_WEBHOOK_SECRET", "liquidity-test-key") + monkeypatch.setenv("VAULTS_MOMMY_WEBHOOK_SECRET", "mommy-test-key") + clock = [1000.0] + monkeypatch.setattr("utils.dispatch_queue.time.time", lambda: clock[0]) + return clock + + +@pytest.fixture +def post(monkeypatch: pytest.MonkeyPatch) -> Mock: + mock = Mock(return_value=SimpleNamespace(status_code=200, headers={}, close=Mock())) + monkeypatch.setattr(dispatch.requests, "post", mock) + return mock + + +def alert(severity: AlertSeverity = AlertSeverity.HIGH, protocol: str = "3jane") -> Alert: + return Alert(severity, "Reserves low — USD3", protocol, channel="pegs") + + +def test_fanout_is_durable_before_first_send_and_keeps_liquidity_envelope(configured: list[float], post: Mock) -> None: + def accept(url: str, **kwargs): + records = DispatchQueue().records() + assert len(records) == 2 and all(row["attempts"] <= 1 for row in records) + return SimpleNamespace(status_code=200, headers={}, close=Mock()) + + post.side_effect = accept + dispatch.dispatch_emergency_withdrawal(alert()) + assert post.call_count == 2 + liquidity, mommy = [call.kwargs for call in post.call_args_list] + legacy = json.loads(liquidity["data"]) + assert legacy == { + "event_type": "emergency_withdrawal", + "client_payload": {"protocol": "3jane", "severity": "HIGH", "message": "Reserves low — USD3"}, + } + new = json.loads(mommy["data"]) + assert new["client_payload"] == legacy["client_payload"] + assert new["event_id"] == DispatchQueue().records()[0]["event_id"] + for kwargs, secret in ((liquidity, "liquidity-test-key"), (mommy, "mommy-test-key")): + assert ( + kwargs["headers"]["X-Hub-Signature-256"] + == "sha256=" + hmac.new(secret.encode(), kwargs["data"], hashlib.sha256).hexdigest() + ) + assert kwargs["allow_redirects"] is False + assert kwargs["timeout"] == (3, 10) + assert all(row["status"] == "delivered" for row in DispatchQueue().records()) + + +def test_high_then_critical_twenty_minutes_later_bypasses_cooldown(configured: list[float], post: Mock) -> None: + dispatch.dispatch_emergency_withdrawal(alert()) + first_ids = {row["event_id"] for row in DispatchQueue().records()} + configured[0] += 1200 + dispatch.dispatch_emergency_withdrawal(alert(AlertSeverity.CRITICAL)) + records = DispatchQueue().records() + assert post.call_count == 4 + assert {row["severity"] for row in records} == {"HIGH", "CRITICAL"} + assert first_ids.isdisjoint({row["event_id"] for row in records if row["severity"] == "CRITICAL"}) + dispatch.dispatch_emergency_withdrawal(alert()) + dispatch.dispatch_emergency_withdrawal(alert(AlertSeverity.CRITICAL)) + assert post.call_count == 4 + configured[0] += 3601 + dispatch.dispatch_emergency_withdrawal(alert(AlertSeverity.CRITICAL)) + assert post.call_count == 6 + + +def test_unavailable_destination_does_not_suppress_other_and_retry_survives_restart( + configured: list[float], post: Mock +) -> None: + def fail_one(url: str, **kwargs): + if "8081" in url: + raise requests.Timeout("down") + return SimpleNamespace(status_code=200, headers={}, close=Mock()) + + post.side_effect = fail_one + dispatch.dispatch_emergency_withdrawal(alert()) + records = DispatchQueue().records() + assert [row["status"] for row in records] == ["delivered", "retry"] + saved_body = post.call_args_list[1].kwargs["data"] + configured[0] += 60 + post.side_effect = None + # A new queue instance models restart. No new alert is generated. + assert dispatch.drain_emergency_deliveries(queue=DispatchQueue()) == 1 + assert post.call_args.kwargs["data"] == saved_body + assert all(row["status"] == "delivered" for row in DispatchQueue().records()) + + +def test_pending_high_never_prevents_new_critical(configured: list[float], post: Mock) -> None: + post.side_effect = requests.ConnectionError("down") + dispatch.dispatch_emergency_withdrawal(alert()) + configured[0] += 1200 + dispatch.dispatch_emergency_withdrawal(alert(AlertSeverity.CRITICAL)) + records = DispatchQueue().records() + assert len(records) == 4 + assert len({row["event_id"] for row in records}) == 2 + + +@pytest.mark.parametrize("code", [408, 429, 500, 503]) +def test_retryable_http_outcomes_keep_stable_id_and_body(configured: list[float], post: Mock, code: int) -> None: + post.return_value = SimpleNamespace(status_code=code, headers={"Retry-After": "120"}, close=Mock()) + dispatch.dispatch_emergency_withdrawal(alert()) + records = DispatchQueue().records() + assert all(row["status"] == "retry" for row in records) + assert all(row["next_attempt_at"] == 1120 if code == 429 else row["next_attempt_at"] == 1030 for row in records) + assert dispatch.drain_emergency_deliveries() == 0 + configured[0] = 1121 + before = [call.kwargs["data"] for call in post.call_args_list] + dispatch.drain_emergency_deliveries() + assert [call.kwargs["data"] for call in post.call_args_list[2:]] == before + + +@pytest.mark.parametrize("code", [302, 400, 401, 403, 404, 422]) +def test_terminal_errors_require_explicit_operator_retry(configured: list[float], post: Mock, code: int) -> None: + post.return_value = SimpleNamespace(status_code=code, headers={}, close=Mock()) + dispatch.dispatch_emergency_withdrawal(alert()) + records = DispatchQueue().records() + assert all(row["status"] == "terminal" for row in records) + configured[0] += 7200 + assert dispatch.drain_emergency_deliveries() == 0 + dispatch.dispatch_emergency_withdrawal(alert()) + assert len(DispatchQueue().records()) == 2 + queue = DispatchQueue() + queue.retry_terminal(records[0]["id"]) + post.return_value = SimpleNamespace(status_code=200, headers={}, close=Mock()) + assert dispatch.drain_emergency_deliveries(queue=queue) == 1 + assert queue.records()[0]["event_id"] == records[0]["event_id"] + + +def test_independent_credentials_and_opt_in_preserve_liquidity( + configured: list[float], post: Mock, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv("VAULTS_MOMMY_WEBHOOK_SECRET") + dispatch.dispatch_emergency_withdrawal(alert()) + assert post.call_count == 1 + assert post.call_args.args[0] == dispatch.DEFAULT_WEBHOOK_URL + monkeypatch.setenv("VAULTS_MOMMY_WEBHOOK_SECRET", "mommy-test-key") + dispatch.dispatch_emergency_withdrawal(alert()) + assert post.call_count == 2 and "8081" in post.call_args.args[0] + + +def test_endpoint_override_remains_compatible( + configured: list[float], post: Mock, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("LIQUIDITY_WEBHOOK_URL", "http://localhost:9000/webhook/emergency") + dispatch.dispatch_emergency_withdrawal(alert()) + assert post.call_args_list[0].args[0] == "http://localhost:9000/webhook/emergency" + + +def test_endpoint_change_needs_explicit_review_before_rerouting( + configured: list[float], post: Mock, monkeypatch: pytest.MonkeyPatch +) -> None: + post.side_effect = requests.Timeout("down") + dispatch.dispatch_emergency_withdrawal(alert()) + configured[0] += 60 + monkeypatch.setenv("VAULTS_MOMMY_WEBHOOK_URL", "http://127.0.0.1:9001/webhook/emergency") + dispatch.drain_emergency_deliveries() + record = DispatchQueue().records()[1] + assert record["status"] == "terminal" + DispatchQueue().retry_terminal(record["id"], url="http://127.0.0.1:9001/webhook/emergency") + post.side_effect = None + dispatch.drain_emergency_deliveries() + assert post.call_args.args[0] == "http://127.0.0.1:9001/webhook/emergency" + + +def test_debug_preserves_existing_queue_without_sending( + configured: list[float], post: Mock, monkeypatch: pytest.MonkeyPatch +) -> None: + queue = DispatchQueue() + queue.enqueue("3jane", "HIGH", "old event", dispatch.destinations()) + before = queue.records() + monkeypatch.setenv("LOG_LEVEL", "DEBUG") + dispatch.dispatch_emergency_withdrawal(alert(AlertSeverity.CRITICAL)) + assert dispatch.drain_emergency_deliveries(queue=queue) == 0 + assert queue.records() == before + post.assert_not_called() + + +@pytest.mark.parametrize( + "severity,protocol", + [(AlertSeverity.LOW, "3jane"), (AlertSeverity.MEDIUM, "3jane"), (AlertSeverity.HIGH, "unknown")], +) +def test_ineligible_alerts_do_not_enqueue_or_send( + configured: list[float], post: Mock, severity: AlertSeverity, protocol: str +) -> None: + dispatch.dispatch_emergency_withdrawal(alert(severity, protocol)) + assert not DispatchQueue().records() + post.assert_not_called() + + +def test_missing_all_secrets_disables_new_delivery( + configured: list[float], post: Mock, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv("LIQUIDITY_WEBHOOK_SECRET") + monkeypatch.delenv("VAULTS_MOMMY_WEBHOOK_SECRET") + dispatch.dispatch_emergency_withdrawal(alert()) + post.assert_not_called() + + +def test_enqueue_disk_failure_prevents_any_http( + configured: list[float], post: Mock, monkeypatch: pytest.MonkeyPatch +) -> None: + def fail(*args, **kwargs): + raise sqlite3.OperationalError("disk full") + + monkeypatch.setattr(DispatchQueue, "enqueue", fail) + with pytest.raises(sqlite3.OperationalError): + dispatch.dispatch_emergency_withdrawal(alert()) + post.assert_not_called() + + +def test_expired_lease_recovers_same_event_and_stale_ack_cannot_clear_it(tmp_path: Path) -> None: + queue = DispatchQueue(tmp_path / "queue.sqlite3") + queue.enqueue("3jane", "HIGH", "test", [Destination("vaults-mommy", "http://local", "secret", True)], now=1000) + first = queue.claim("vaults-mommy", exclude=set(), now=1000) + assert first is not None + assert queue.claim("vaults-mommy", exclude=set(), now=1001) is None + restarted = DispatchQueue(tmp_path / "queue.sqlite3") + next_attempt = restarted.claim("vaults-mommy", exclude=set(), now=1061) + assert next_attempt is not None + assert next_attempt.body == first.body and next_attempt.event_id == first.event_id + with pytest.raises(RuntimeError, match="lease changed"): + queue.finish(first, "delivered", now=1062) + restarted.finish(next_attempt, "delivered", now=1062) + assert restarted.records()[0]["attempts"] == 2 + + +def _enqueue_concurrent(path: str) -> None: + DispatchQueue(path).enqueue( + "3jane", "HIGH", "same incident", [Destination("liquidity", "http://local", "key")], now=1000 + ) + + +def test_concurrent_monitors_do_not_duplicate_pending_protocol(tmp_path: Path) -> None: + path = str(tmp_path / "queue.sqlite3") + DispatchQueue(path).records() + ctx = multiprocessing.get_context("fork") + writers = [ctx.Process(target=_enqueue_concurrent, args=(path,)) for _ in range(8)] + for process in writers: + process.start() + for process in writers: + process.join(timeout=10) + assert process.exitcode == 0 + assert len(DispatchQueue(path).records()) == 1 + + +def test_unknown_database_schema_refuses_delivery(tmp_path: Path) -> None: + path = tmp_path / "queue.sqlite3" + with sqlite3.connect(path) as conn: + conn.execute("PRAGMA user_version=999") + with pytest.raises(ValueError, match="incompatible"): + DispatchQueue(path).records() + + +def test_retry_job_is_scheduled_without_monitor_alerts() -> None: + jobs = load_jobs_config() + profile = jobs.profiles["emergency_delivery"] + assert profile.cron == "* * * * *" + assert any(task.script == "utils/dispatch_retry.py" for task in profile.enabled_tasks) + + +def test_successful_http_with_failed_ack_persistence_remains_uncertain( + configured: list[float], post: Mock, monkeypatch: pytest.MonkeyPatch +) -> None: + original = DispatchQueue.finish + + def fail(self, delivery, status, **kwargs): + raise sqlite3.OperationalError("ack commit failed") + + monkeypatch.setattr(DispatchQueue, "finish", fail) + with pytest.raises(sqlite3.OperationalError): + dispatch.dispatch_emergency_withdrawal(alert()) + first = post.call_args.kwargs["data"] + monkeypatch.setattr(DispatchQueue, "finish", original) + configured[0] += 61 + dispatch.drain_emergency_deliveries(queue=DispatchQueue()) + assert post.call_args_list[1].kwargs["data"] == first + assert all(row["status"] == "delivered" for row in DispatchQueue().records()) + + +def test_configured_secrets_are_not_stored(configured: list[float], post: Mock) -> None: + queue = DispatchQueue() + dispatch.dispatch_emergency_withdrawal(alert()) + for path in Path(queue.path).parent.glob(Path(queue.path).name + "*"): + assert b"liquidity-test-key" not in path.read_bytes() + assert b"mommy-test-key" not in path.read_bytes() diff --git a/utils/dispatch.md b/utils/dispatch.md new file mode 100644 index 00000000..893aec76 --- /dev/null +++ b/utils/dispatch.md @@ -0,0 +1,59 @@ +# Durable emergency delivery + +`utils.dispatch` queues HIGH/CRITICAL monitoring alerts before sending them to +independently configured liquidity-monitoring and vaults-mommy destinations. +`automation/jobs.yaml` drains retries every minute, even without a new alert. + +## Destinations and operations + + +Set `VAULTS_MOMMY_WEBHOOK_SECRET` independently from `LIQUIDITY_WEBHOOK_SECRET`. +Its default URL is `http://127.0.0.1:8081/webhook/emergency`; override with +`VAULTS_MOMMY_WEBHOOK_URL`. Leaving either secret unset disables only that destination. +Receiver execution/position activation remains a separate operator decision under +[vaults-mommy #27/#35](https://github.com/johnnyonline/vaults-mommy/issues/27). +This sender neither signs transactions nor approves an exit. + +Before HTTP, eligible deliveries for both destinations are committed to +`$CACHE_DIR/emergency_dispatch.sqlite3` (`EMERGENCY_DISPATCH_DB` overrides it). +SQLite uses WAL, FULL synchronization, transactions and bounded lock waits. +Back up the database and its WAL with the other monitoring state; do not delete it to +clear a failure. It stores payloads, endpoint URLs, lease/attempt state and independent +success timestamps, never webhook secrets. New code starts a new cooldown history; +legacy `dispatch_last_` cache values are retained but are not consulted. +The first alert after upgrading can therefore be delivered again. + +Liquidity receives the existing `event_type`/`client_payload` envelope unchanged. +Vaults-mommy additionally receives top-level `event_id` inside the signed body. +The ID and exact UTF-8 payload bytes remain stable across retries and process restarts. +An escalation is a new logical event/ID. HIGH at t=0 followed by CRITICAL at t=20 minutes +is delivered to both independently. Neither HIGH success nor HIGH pending/terminal work +suppresses CRITICAL. An unavailable destination cannot mark another as delivered or +start its cooldown. Duplicate same-severity pending work is retained, not multiplied. + +The `emergency_delivery` automation profile runs `utils/dispatch_retry.py` every minute, +so retries drain even without new alerts. Timeout/connection failures, HTTP 408, 429 +and 5xx retry with bounded backoff; numeric Retry-After is honored up to an hour. +3xx redirects are not followed. Other 4xx outcomes remain terminal and visible in logs +for correction. Repeated alerts do not silently replace terminal deliveries. +Leases expire after a crash, allowing at-least-once redelivery with the same event ID. +If an HTTP acknowledgement cannot be committed, delivery remains uncertain and retries. + +After correcting a terminal error, explicitly retry its logged delivery ID: + +```sh +uv run python -m utils.dispatch_retry --retry-terminal +``` + +This preserves ID and payload. It also explicitly adopts the destination's current URL, +if the endpoint was corrected. An endpoint change alone never reroutes previously queued +signed alerts automatically. Secret rotation uses the current secret on the same body. +Missing/removed credentials hold existing pending deliveries; scheduled logs identify them. +DEBUG skips enqueue and sending while retaining queued state. The script uses the existing +crash-alert wrapper; disk/schema failures are operational errors, never delivery success. + +The vaults-mommy #29 receiver integration must continue accepting legacy manual/no-ID envelopes. Without a +stable ID, an authentic delayed/replayed alert can reopen a resolved incident and request +execution; ordinary readiness, execution-gate and loss checks still apply. No timestamp or +resolution freshness gate is invented. Do not turn on automatic CRITICAL execution until +the sender escalation fix and receiver acceptance tests pass together. diff --git a/utils/dispatch.py b/utils/dispatch.py index 587257c7..e847fcb8 100644 --- a/utils/dispatch.py +++ b/utils/dispatch.py @@ -1,34 +1,23 @@ -"""Dispatch emergency withdrawal requests to the liquidity-monitoring webhook. +"""Durably deliver risk alerts to liquidity-monitoring and optional vaults-mommy. -When a HIGH or CRITICAL alert fires, this module sends a signed webhook request -to the liquidity-monitoring service with the protocol name and severity. The -receiving service resolves which vaults/markets to act on from its own config -(``emergency_config.json``, ``markets_config.py``, ``forced_caps.json``). - -Requires the ``LIQUIDITY_WEBHOOK_SECRET`` environment variable. +The existing liquidity envelope is preserved byte-for-byte in shape; vaults-mommy +also receives a signed event_id. Retry bodies and IDs survive process restarts. """ import hashlib import hmac -import json import os -import time import requests from utils.alert import Alert, AlertSeverity -from utils.cache import cache_filename, get_last_value_for_key_from_file, write_last_value_to_file +from utils.dispatch_queue import Destination, DispatchQueue from utils.logger import get_logger logger = get_logger("utils.dispatch") - DEFAULT_WEBHOOK_URL = "http://127.0.0.1:8080/webhook/emergency" -WEBHOOK_URL_ENV = "LIQUIDITY_WEBHOOK_URL" -WEBHOOK_SECRET_ENV = "LIQUIDITY_WEBHOOK_SECRET" -DEFAULT_COOLDOWN_SECONDS = 3600 # 60 minutes +DEFAULT_VAULTS_MOMMY_URL = "http://127.0.0.1:8081/webhook/emergency" -# Protocols that have emergency withdrawal config in liquidity-monitoring. -# Only these protocols will trigger a dispatch. DISPATCHABLE_PROTOCOLS = { "infinifi", "cap", @@ -47,97 +36,122 @@ } -def _is_on_cooldown(protocol: str, cooldown_seconds: int = DEFAULT_COOLDOWN_SECONDS) -> bool: - """Check if a dispatch was sent recently for this protocol.""" - cache_key = f"dispatch_last_{protocol}" - last_ts = get_last_value_for_key_from_file(cache_filename, cache_key) - if last_ts == 0: - return False - try: - return (time.time() - float(last_ts)) < cooldown_seconds - except TypeError, ValueError: - return False - - -def _record_dispatch(protocol: str) -> None: - """Record the current timestamp as the last dispatch time for this protocol.""" - cache_key = f"dispatch_last_{protocol}" - write_last_value_to_file(cache_filename, cache_key, time.time()) - - -def _serialize_payload(payload: dict) -> bytes: - """Serialize once so the signed bytes are exactly the bytes sent.""" - return json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8") +def destinations() -> tuple[Destination, ...]: + """Load independent credentials; an unset secret disables only its destination.""" + result = [] + for name, prefix, default, signed_id in ( + ("liquidity", "LIQUIDITY_WEBHOOK", DEFAULT_WEBHOOK_URL, False), + ("vaults-mommy", "VAULTS_MOMMY_WEBHOOK", DEFAULT_VAULTS_MOMMY_URL, True), + ): + secret = os.getenv(f"{prefix}_SECRET") + if secret: + result.append(Destination(name, os.getenv(f"{prefix}_URL", default), secret, signed_id)) + elif os.getenv(f"{prefix}_URL"): + logger.warning("%s_SECRET missing; %s dispatch is disabled", prefix, name) + return tuple(result) def _signature_header(secret: str, body: bytes) -> str: - digest = hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest() - return f"sha256={digest}" - + """Sign the exact persisted bytes that will be sent.""" + return "sha256=" + hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest() -def dispatch_emergency_withdrawal(alert: Alert) -> None: - """Dispatch an emergency withdrawal to liquidity-monitoring. - Registered by default when ``utils.alert`` is imported (see - ``_ensure_default_dispatch_hook``) unless a hook was already set; override via - ``register_alert_hook``. - Only dispatches for HIGH and CRITICAL alerts whose protocol is in - ``DISPATCHABLE_PROTOCOLS``. Respects a per-protocol cooldown to avoid - duplicate dispatches from repeated alerts. +def drain_emergency_deliveries(*, queue: DispatchQueue | None = None, limit: int = 20) -> int: + """Drain due work without needing a new alert; each destination progresses independently.""" + if os.getenv("LOG_LEVEL", "INFO").upper() == "DEBUG": + logger.debug("Skipping emergency delivery retries (LOG_LEVEL=DEBUG); queue retained") + return 0 + configured = destinations() + if not configured: + return 0 + queue = queue or DispatchQueue() + attempted: set[int] = set() + for _ in range(limit): + progressed = False + for destination in configured: + delivery = queue.claim(destination.name, exclude=attempted) + if delivery is None: + continue + progressed = True + attempted.add(delivery.delivery_id) + if destination.url != delivery.url: + queue.finish( + delivery, "terminal", error="configured endpoint changed; explicitly retry after operator review" + ) + logger.error("Emergency delivery %s endpoint changed; retained for review", delivery.delivery_id) + continue + try: + response = requests.post( + delivery.url, + data=delivery.body, + headers={ + "Content-Type": "application/json", + "X-Hub-Signature-256": _signature_header(destination.secret, delivery.body), + }, + timeout=(3, 10), + allow_redirects=False, + stream=True, + ) + try: + code = response.status_code + delay = None + if code == 429: + raw_delay = response.headers.get("Retry-After", "") + if raw_delay.isdecimal(): + delay = min(3600, int(raw_delay)) + finally: + response.close() + except requests.RequestException as error: + queue.finish(delivery, "retry", error=type(error).__name__) + logger.warning( + "Emergency delivery %s to %s failed (%s); retry retained", + delivery.delivery_id, + destination.name, + type(error).__name__, + ) + continue + if 200 <= code < 300: + queue.finish(delivery, "delivered") + logger.info( + "Emergency event %s delivered to %s (%s)", delivery.event_id, destination.name, delivery.severity + ) + elif code in {408, 429} or 500 <= code < 600: + queue.finish(delivery, "retry", error=f"HTTP {code}", retry_after=delay) + logger.warning( + "Emergency delivery %s to %s returned %s; retry retained", + delivery.delivery_id, + destination.name, + code, + ) + else: + queue.finish( + delivery, "terminal", error=f"HTTP {code}; correct configuration/payload then explicitly retry" + ) + logger.error( + "Emergency delivery %s to %s terminal HTTP %s; operator correction required", + delivery.delivery_id, + destination.name, + code, + ) + if not progressed: + break + return len(attempted) - The receiving webhook resolves vaults, markets, and chains from its own - ``emergency_config.json``. - Args: - alert: The alert that triggered the hook. - """ - if alert.severity not in (AlertSeverity.HIGH, AlertSeverity.CRITICAL): +def dispatch_emergency_withdrawal(alert: Alert) -> None: + """Persist fanout before sending; CRITICAL bypasses HIGH's destination cooldown.""" + if ( + alert.severity not in (AlertSeverity.HIGH, AlertSeverity.CRITICAL) + or alert.protocol not in DISPATCHABLE_PROTOCOLS + ): return - if os.getenv("LOG_LEVEL", "INFO").upper() == "DEBUG": logger.debug("Skipping dispatch (LOG_LEVEL=DEBUG)") return - - if alert.protocol not in DISPATCHABLE_PROTOCOLS: - logger.debug("Protocol %s not in DISPATCHABLE_PROTOCOLS, skipping dispatch", alert.protocol) - return - - if _is_on_cooldown(alert.protocol): - logger.info("Dispatch for %s is on cooldown, skipping", alert.protocol) + configured = destinations() + if not configured: + logger.warning("No emergency webhook secrets configured; dispatch disabled") return - - secret = os.getenv(WEBHOOK_SECRET_ENV) - if not secret: - logger.warning("%s not set, cannot dispatch emergency withdrawal", WEBHOOK_SECRET_ENV) - return - - payload = { - "event_type": "emergency_withdrawal", - "client_payload": { - "protocol": alert.protocol, - "severity": alert.severity.value, - "message": alert.message, - }, - } - body = _serialize_payload(payload) - webhook_url = os.getenv(WEBHOOK_URL_ENV, DEFAULT_WEBHOOK_URL) - - try: - response = requests.post( - webhook_url, - data=body, - headers={ - "Content-Type": "application/json", - "X-Hub-Signature-256": _signature_header(secret, body), - }, - timeout=10, - ) - response.raise_for_status() - _record_dispatch(alert.protocol) - logger.info( - "Dispatched emergency withdrawal for %s (severity=%s)", - alert.protocol, - alert.severity.value, - ) - except requests.RequestException: - logger.exception("Failed to dispatch emergency withdrawal for %s", alert.protocol) + queue = DispatchQueue() + queue.enqueue(alert.protocol, alert.severity.value, alert.message, configured) + drain_emergency_deliveries(queue=queue) diff --git a/utils/dispatch_queue.py b/utils/dispatch_queue.py new file mode 100644 index 00000000..04babb54 --- /dev/null +++ b/utils/dispatch_queue.py @@ -0,0 +1,228 @@ +"""Durable emergency fanout with independent destination success and retry state. + +SQLite FULL commits precede every HTTP attempt. Leased attempts can be retried +following process failure, so delivery is at least once; receivers must dedupe. +Secrets are resolved at send time and never written to the queue. +""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import time +import uuid +from collections.abc import Iterator, Sequence +from contextlib import closing, contextmanager +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from utils.paths import cache_path + +SCHEMA_VERSION = 1 +COOLDOWN_SECONDS = 3600 +LEASE_SECONDS = 60 + + +@dataclass(frozen=True) +class Destination: + """A configured endpoint; credentials stay in memory.""" + + name: str + url: str + secret: str = field(repr=False) + signed_event_id: bool = False + + +@dataclass(frozen=True) +class Delivery: + """One independently tracked destination of a logical alert.""" + + delivery_id: int + event_id: str + destination: str + url: str + body: bytes + protocol: str + severity: str + attempts: int + lease_token: str + + +class DispatchQueue: + """Cross-process transactional queue; no network I/O occurs inside a DB lock.""" + + def __init__(self, path: str | Path | None = None) -> None: + self.path = str(path or os.getenv("EMERGENCY_DISPATCH_DB") or cache_path("emergency_dispatch.sqlite3")) + + @contextmanager + def _connect(self) -> Iterator[sqlite3.Connection]: + with closing(sqlite3.connect(self.path, timeout=5, isolation_level=None)) as conn: + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=FULL") + conn.execute("PRAGMA foreign_keys=ON") + conn.execute("BEGIN IMMEDIATE") + try: + version = conn.execute("PRAGMA user_version").fetchone()[0] + if version not in (0, SCHEMA_VERSION): + raise ValueError(f"incompatible emergency dispatch DB version: {version}") + if version == 0: + for statement in _SCHEMA: + conn.execute(statement) + conn.execute(f"PRAGMA user_version={SCHEMA_VERSION}") + conn.commit() + yield conn + except BaseException: + conn.rollback() + raise + + def enqueue( + self, + protocol: str, + severity: str, + message: str, + destinations: Sequence[Destination], + *, + now: float | None = None, + ) -> str | None: + """Persist eligible destinations atomically, preserving a distinct escalation ID. + + HIGH is suppressed by recent HIGH/CRITICAL success or pending work. + CRITICAL is suppressed only by CRITICAL success/pending work, never HIGH. + Failures at one destination cannot suppress another destination's event. + """ + if severity not in {"HIGH", "CRITICAL"}: + raise ValueError("only HIGH/CRITICAL can be dispatched") + timestamp = time.time() if now is None else now + event_id = uuid.uuid4().hex + with self._connect() as conn, conn: + conn.execute("BEGIN IMMEDIATE") + selected = [] + for destination in destinations: + relevant = ("HIGH", "CRITICAL") if severity == "HIGH" else ("CRITICAL",) + placeholders = ",".join("?" for _ in relevant) + success = conn.execute( + f"SELECT 1 FROM successes WHERE destination=? AND protocol=? AND severity IN ({placeholders}) AND succeeded_at>? LIMIT 1", + (destination.name, protocol, *relevant, timestamp - COOLDOWN_SECONDS), + ).fetchone() + pending = conn.execute( + f"SELECT 1 FROM deliveries WHERE destination=? AND protocol=? AND severity IN ({placeholders}) AND status IN ('pending','retry','terminal') LIMIT 1", + (destination.name, protocol, *relevant), + ).fetchone() + if not success and not pending: + selected.append(destination) + if not selected: + return None + for destination in selected: + payload: dict[str, Any] = { + "event_type": "emergency_withdrawal", + "client_payload": {"protocol": protocol, "severity": severity, "message": message}, + } + if destination.signed_event_id: + payload["event_id"] = event_id + body = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + conn.execute( + "INSERT INTO deliveries(event_id,destination,url,body,protocol,severity,created_at,next_attempt_at) VALUES (?,?,?,?,?,?,?,?)", + (event_id, destination.name, destination.url, body, protocol, severity, timestamp, timestamp), + ) + return event_id + + def claim(self, destination: str, *, exclude: set[int], now: float | None = None) -> Delivery | None: + """Commit an expiring lease before releasing the DB lock for HTTP.""" + timestamp = time.time() if now is None else now + token = uuid.uuid4().hex + exclusions = " AND id NOT IN (" + ",".join("?" for _ in exclude) + ")" if exclude else "" + with self._connect() as conn, conn: + conn.execute("BEGIN IMMEDIATE") + row = conn.execute( + "SELECT * FROM deliveries WHERE destination=? AND status IN ('pending','retry') AND next_attempt_at<=? AND lease_until<=?" + + exclusions + + " ORDER BY CASE severity WHEN 'CRITICAL' THEN 0 ELSE 1 END, created_at, id LIMIT 1", + (destination, timestamp, timestamp, *sorted(exclude)), + ).fetchone() + if row is None: + return None + conn.execute( + "UPDATE deliveries SET lease_until=?,lease_token=?,attempts=attempts+1 WHERE id=?", + (timestamp + LEASE_SECONDS, token, row["id"]), + ) + return Delivery( + row["id"], + row["event_id"], + row["destination"], + row["url"], + bytes(row["body"]), + row["protocol"], + row["severity"], + row["attempts"] + 1, + token, + ) + + def finish( + self, + delivery: Delivery, + status: str, + *, + error: str | None = None, + retry_after: float | None = None, + now: float | None = None, + ) -> None: + """Commit acknowledgement or failure; a stale lease cannot overwrite a new one.""" + if status not in {"delivered", "retry", "terminal"}: + raise ValueError("invalid delivery outcome") + timestamp = time.time() if now is None else now + delay = min(1800, 30 * 2 ** min(delivery.attempts - 1, 6)) if retry_after is None else retry_after + with self._connect() as conn, conn: + conn.execute("BEGIN IMMEDIATE") + cursor = conn.execute( + "UPDATE deliveries SET status=?,last_error=?,next_attempt_at=?,lease_until=0,lease_token=NULL WHERE id=? AND lease_token=?", + (status, error, timestamp + delay, delivery.delivery_id, delivery.lease_token), + ) + if cursor.rowcount != 1: + raise RuntimeError("emergency delivery lease changed; acknowledgement remains uncertain") + if status == "delivered": + conn.execute( + "INSERT INTO successes(destination,protocol,severity,succeeded_at) VALUES(?,?,?,?) ON CONFLICT(destination,protocol,severity) DO UPDATE SET succeeded_at=MAX(succeeded_at,excluded.succeeded_at)", + (delivery.destination, delivery.protocol, delivery.severity, timestamp), + ) + + def records(self) -> list[dict[str, Any]]: + """Inspect delivery state, never credentials or arbitrary response bodies.""" + with self._connect() as conn: + return [ + dict(row) + for row in conn.execute( + "SELECT id,event_id,destination,protocol,severity,status,attempts,next_attempt_at,last_error FROM deliveries ORDER BY id" + ) + ] + + def retry_terminal(self, delivery_id: int, *, url: str | None = None) -> None: + """Explicit operator retry after correcting a terminal error; preserve event ID/body.""" + with self._connect() as conn, conn: + conn.execute("BEGIN IMMEDIATE") + cursor = conn.execute( + "UPDATE deliveries SET status='retry',url=COALESCE(?,url),next_attempt_at=0,last_error=NULL WHERE id=? AND status='terminal'", + (url, delivery_id), + ) + if cursor.rowcount != 1: + raise ValueError("delivery is not a terminal failure") + + +_SCHEMA = ( + """CREATE TABLE deliveries ( + id INTEGER PRIMARY KEY, event_id TEXT NOT NULL, destination TEXT NOT NULL, + url TEXT NOT NULL, body BLOB NOT NULL, protocol TEXT NOT NULL, + severity TEXT NOT NULL CHECK(severity IN ('HIGH','CRITICAL')), + status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending','retry','delivered','terminal')), + created_at REAL NOT NULL, next_attempt_at REAL NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, lease_until REAL NOT NULL DEFAULT 0, + lease_token TEXT, last_error TEXT, UNIQUE(event_id,destination) + )""", + "CREATE INDEX deliveries_due ON deliveries(destination,status,next_attempt_at)", + """CREATE TABLE successes ( + destination TEXT NOT NULL, protocol TEXT NOT NULL, severity TEXT NOT NULL, + succeeded_at REAL NOT NULL, PRIMARY KEY(destination,protocol,severity) + )""", +) diff --git a/utils/dispatch_retry.py b/utils/dispatch_retry.py new file mode 100644 index 00000000..0abd0d79 --- /dev/null +++ b/utils/dispatch_retry.py @@ -0,0 +1,45 @@ +"""Scheduled emergency retries, independent of newly generated monitoring alerts.""" + +import argparse + +from utils.dispatch import destinations, drain_emergency_deliveries +from utils.dispatch_queue import DispatchQueue +from utils.logger import get_logger + +logger = get_logger("utils.dispatch_retry") + + +def main() -> None: + """Retry due deliveries; expose retained terminal failures for operator correction.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--retry-terminal", type=int, help="Explicitly retry a corrected terminal delivery, preserving its ID" + ) + args = parser.parse_args() + queue = DispatchQueue() + if args.retry_terminal is not None: + record = next((row for row in queue.records() if row["id"] == args.retry_terminal), None) + target = next((item for item in destinations() if record and item.name == record["destination"]), None) + if target is None: + raise ValueError("configure the delivery destination before explicitly retrying") + queue.retry_terminal(args.retry_terminal, url=target.url) + count = drain_emergency_deliveries(queue=queue) + records = queue.records() + terminal = [row for row in records if row["status"] == "terminal"] + configured = {item.name for item in destinations()} + for row in records: + if row["status"] in {"pending", "retry"} and row["destination"] not in configured: + logger.warning( + "Emergency delivery %s held: %s credentials are not configured", row["id"], row["destination"] + ) + logger.info("Emergency outbox: attempted %s deliveries; %s terminal failures", count, len(terminal)) + for row in terminal: + logger.warning( + "Emergency delivery %s to %s needs correction: %s", row["id"], row["destination"], row["last_error"] + ) + + +if __name__ == "__main__": + from utils.runner import run_with_alert + + run_with_alert(main, "yearn", alert_protocol="yearn-internal")