From d15ada6396ad4c6812da20fedeb7dec27b5a5a31 Mon Sep 17 00:00:00 2001 From: Basil Zafar Date: Wed, 11 Nov 2026 23:37:22 +0530 Subject: [PATCH] fix(cli): honor bounded Retry-After for Hosted ingest 429s --- docs/CLI.md | 8 + docs/HOSTED_INGESTION.md | 25 ++- .../mutiny_cli/src/mutiny_cli/hosted_sync.py | 67 ++++--- tests/unit/test_cli_hosted_sync.py | 180 ++++++++++++++++++ 4 files changed, 254 insertions(+), 26 deletions(-) diff --git a/docs/CLI.md b/docs/CLI.md index cd2a6c0..8a6e327 100644 --- a/docs/CLI.md +++ b/docs/CLI.md @@ -89,6 +89,14 @@ the freed name; if no finding reproduces, restore the backup. `mutiny run` has n **Hosted sync (M-PR8C):** After local campaign completion, CLI POSTs to `/api/ingest/v1/*` with `schema_version=1` and `redaction.applied=true`. Bearer from `MUTINY_API_TOKEN` when set. Local success + sync failure → exit **3** (local result remains authoritative). Missing `api_url` with `--hosted` → exit 2. Config URL alone never selects Hosted. +**Rate limits:** Each ingest POST retries HTTP `429` up to twice (three attempts +total), reusing the identical sanitized payload and IDs. `Retry-After` may be +integer seconds or an HTTP date; each delay is capped at five seconds, with a +one-second fallback for missing/malformed headers and zero for elapsed delays. +Other HTTP errors and transport failures are not retried. After exhaustion, the +existing sync failure/exit behavior and sanitized pending file are preserved. +See [ingest retry semantics](./HOSTED_INGESTION.md#bounded-rate-limit-retries-43). + **Destination policy:** By default, Hosted sync accepts only HTTP(S) URLs whose host is `localhost` or a literal loopback IP (`127.0.0.0/8` or `::1`). Other hosts require `--allow-remote-hosted` on the command line, even with `--hosted-url` or diff --git a/docs/HOSTED_INGESTION.md b/docs/HOSTED_INGESTION.md index d7e9109..9145d77 100644 --- a/docs/HOSTED_INGESTION.md +++ b/docs/HOSTED_INGESTION.md @@ -250,7 +250,7 @@ CLI still reports local success/failure as today ↓ warn on stderr; write .mutiny/hosted-pending/.json (sanitized) ↓ -retry manually; `mutiny sync` does not exist — no automatic retry in M-PR8C +manual recovery after bounded 429 retries; `mutiny sync` does not exist ``` **Rules:** @@ -261,6 +261,27 @@ retry manually; `mutiny sync` does not exist — no automatic retry in M-PR8C 4. No Hosted round-trip inside Core evaluation loops. 5. `MUTINY_DISABLE_SECRET_REDACTION=1` refuses Hosted upload (sync exit 3 after local success); local persistence still allowed. +### Bounded rate-limit retries (#43) + +The CLI retries each idempotent ingest POST only on HTTP `429`: at most **two +retries** after the initial attempt (**three attempts total**). Campaign open, +event/artifact batches, completion, regressions, and the test-run client method +share this behavior. The request's already-redacted payload and identifiers are +reused unchanged; payloads and timestamps are not regenerated between attempts. + +`Retry-After` accepts integer seconds or an HTTP date. Each delay is clamped to +**0–5 seconds**; past dates and negative delays wait zero seconds. Missing or +malformed headers use **one second**. Scheduled waiting is at most **10 seconds +per POST**, in addition to the existing HTTP timeout for each attempt. No sleep +occurs after the last failed attempt. + +Non-429 responses and transport errors are not retried. Exhausted 429s use the +existing failure path: local results remain authoritative, local success plus +sync failure exits `3`, and sanitized pending JSON is saved when enabled. +Replaying pending files and a `mutiny sync` command remain future work; these +immediate retries do not add a background sync queue. Redaction, destination +validation, bearer authentication, and redirect refusal remain required. + --- ## 12. Versioning @@ -471,5 +492,5 @@ Preserved by M-PR8E (customer Hosted exec retired; demo + ingest retained): ### Current limitations (post M-PR8E) - `mutiny test --hosted` not wired (payload builder only) — remaining CLI sync work / later -- Automatic retry / `mutiny sync` command deferred (pending JSON written on sync failure) +- Pending-file replay / `mutiny sync` command deferred; immediate HTTP 429 retries are bounded as described above (pending JSON written on final sync failure) - Web cannot distinguish local-success + sync-failure from a missing Hosted row: sync failures do not fully ingest, so Hosted never claims those runs were received. A terminal `failed` status on an ingested `execution_mode=local_cli` campaign means the CLI reported campaign failure, not Hosted sync failure. diff --git a/packages/mutiny_cli/src/mutiny_cli/hosted_sync.py b/packages/mutiny_cli/src/mutiny_cli/hosted_sync.py index c6b9f2e..5107a9c 100644 --- a/packages/mutiny_cli/src/mutiny_cli/hosted_sync.py +++ b/packages/mutiny_cli/src/mutiny_cli/hosted_sync.py @@ -14,9 +14,11 @@ import json import logging import os +import time import uuid from dataclasses import dataclass, field from datetime import datetime, timezone +from email.utils import parsedate_to_datetime from pathlib import Path from typing import Any, Literal from urllib.parse import urljoin @@ -31,6 +33,9 @@ INGEST_SCHEMA_VERSION = 1 EXECUTION_MODE = "local_cli" DEFAULT_TIMEOUT_S = 30.0 +MAX_RATE_LIMIT_RETRIES = 2 +MAX_RETRY_AFTER_S = 5.0 +DEFAULT_RETRY_AFTER_S = 1.0 # Local campaign OK but Hosted sync failed / refused (distinct from local fail=1). SYNC_FAILED_EXIT = 3 @@ -484,13 +489,30 @@ def write_pending_bundle(bundle: LocalRunBundle, payloads: dict[str, Any]) -> Pa "kind": "hosted_pending_sync", "campaign_id": bundle.campaign_id, "created_at": _utcnow(), - "note": "Sanitized ingest payloads; not executable. Retry is future work.", + "note": "Sanitized ingest payloads; not executable. Pending-file replay is future work.", "payloads": payloads, } path.write_text(json.dumps(doc, indent=2), encoding="utf-8") return path +def _retry_after_seconds(value: str | None) -> float: + """Parse delta-seconds or an HTTP date, with a bounded fallback delay.""" + if value is None: + return DEFAULT_RETRY_AFTER_S + try: + delay = float(int(value)) + except (ValueError, OverflowError): + try: + when = parsedate_to_datetime(value) + if when.tzinfo is None: + when = when.replace(tzinfo=timezone.utc) + delay = (when - datetime.now(timezone.utc)).total_seconds() + except (ValueError, TypeError, OverflowError): + return DEFAULT_RETRY_AFTER_S + return min(MAX_RETRY_AFTER_S, max(0.0, delay)) + + class HostedIngestClient: """Thin httpx client for ``/api/ingest/v1/*`` (timeouts; no token logging).""" @@ -553,41 +575,38 @@ def _raise_for_response(self, response: Any, *, action: str) -> None: msg = f"{msg}: {detail[:160]}" raise HostedSyncError(msg, error_code=code, http_status=status) - def open_campaign(self, payload: dict[str, Any]) -> dict[str, Any]: + def _post( + self, path: str, payload: dict[str, Any], *, action: str + ) -> dict[str, Any]: with self._client() as client: - r = client.post("/api/ingest/v1/campaigns", json=payload) - self._raise_for_response(r, action="campaign open") + for attempt in range(MAX_RATE_LIMIT_RETRIES + 1): + r = client.post(path, json=payload) + if r.status_code != 429 or attempt == MAX_RATE_LIMIT_RETRIES: + break + time.sleep(_retry_after_seconds(r.headers.get("Retry-After"))) + self._raise_for_response(r, action=action) return r.json() + def open_campaign(self, payload: dict[str, Any]) -> dict[str, Any]: + return self._post("/api/ingest/v1/campaigns", payload, action="campaign open") + def post_batch(self, campaign_id: str, payload: dict[str, Any]) -> dict[str, Any]: - with self._client() as client: - r = client.post( - f"/api/ingest/v1/campaigns/{campaign_id}/batch", json=payload - ) - self._raise_for_response(r, action="batch ingest") - return r.json() + return self._post( + f"/api/ingest/v1/campaigns/{campaign_id}/batch", payload, action="batch ingest" + ) def complete_campaign( self, campaign_id: str, payload: dict[str, Any] ) -> dict[str, Any]: - with self._client() as client: - r = client.post( - f"/api/ingest/v1/campaigns/{campaign_id}/complete", json=payload - ) - self._raise_for_response(r, action="campaign complete") - return r.json() + return self._post( + f"/api/ingest/v1/campaigns/{campaign_id}/complete", payload, action="campaign complete" + ) def post_regression(self, payload: dict[str, Any]) -> dict[str, Any]: - with self._client() as client: - r = client.post("/api/ingest/v1/regressions", json=payload) - self._raise_for_response(r, action="regression ingest") - return r.json() + return self._post("/api/ingest/v1/regressions", payload, action="regression ingest") def post_test_run(self, payload: dict[str, Any]) -> dict[str, Any]: - with self._client() as client: - r = client.post("/api/ingest/v1/test-runs", json=payload) - self._raise_for_response(r, action="test-run ingest") - return r.json() + return self._post("/api/ingest/v1/test-runs", payload, action="test-run ingest") def sync_local_campaign( diff --git a/tests/unit/test_cli_hosted_sync.py b/tests/unit/test_cli_hosted_sync.py index 31ec0b5..36d864a 100644 --- a/tests/unit/test_cli_hosted_sync.py +++ b/tests/unit/test_cli_hosted_sync.py @@ -1078,3 +1078,183 @@ def test_30_sync_module_has_no_exec_gadgets() -> None: "eval(", ): assert needle not in src + + +@pytest.fixture +def rate_limited_client(monkeypatch): + import httpx + + requests = [] + sleeps = [] + responses = [] + client = HostedIngestClient("http://127.0.0.1:8000", headers={"Authorization": f"Bearer {FAKE_TOKEN}"}) + + def handle(request): + requests.append(request) + response = responses.pop(0) + if isinstance(response, Exception): + raise response + return response + + monkeypatch.delenv("MUTINY_DISABLE_SECRET_REDACTION", raising=False) + monkeypatch.setattr(hosted_sync.time, "sleep", sleeps.append) + monkeypatch.setattr(client, "_client", lambda: httpx.Client( + base_url=client.api_url, headers=client.headers, follow_redirects=False, + transport=httpx.MockTransport(handle), + )) + return client, requests, sleeps, responses + + +@pytest.mark.parametrize("method,path", [ + ("open_campaign", "/api/ingest/v1/campaigns"), + ("post_batch", "/api/ingest/v1/campaigns/c1/batch"), + ("complete_campaign", "/api/ingest/v1/campaigns/c1/complete"), + ("post_regression", "/api/ingest/v1/regressions"), + ("post_test_run", "/api/ingest/v1/test-runs"), +]) +def test_ingest_posts_retry_429_with_identical_payload(rate_limited_client, method, path, caplog): + import httpx + + client, requests, sleeps, responses = rate_limited_client + responses.extend([ + httpx.Response(429, headers={"Retry-After": "2"}), + httpx.Response(200, json={"ok": True}), + ]) + payload = {"schema_version": 1, "redaction": {"applied": True}, "event_id": "stable-id"} + args = ("c1", payload) if method in {"post_batch", "complete_campaign"} else (payload,) + assert getattr(client, method)(*args) == {"ok": True} + assert sleeps == [2.0] + assert len(requests) == 2 + assert all(request.url.path == path for request in requests) + assert requests[0].content == requests[1].content + assert all(request.headers["Authorization"] == f"Bearer {FAKE_TOKEN}" for request in requests) + assert FAKE_TOKEN not in caplog.text + + +@pytest.mark.parametrize("value,expected", [ + (None, 1.0), ("", 1.0), ("invalid", 1.0), ("NaN", 1.0), ("inf", 1.0), + ("-10", 0.0), ("0", 0.0), ("999999", 5.0), +]) +def test_retry_delay_is_bounded_with_safe_fallback(rate_limited_client, value, expected): + import httpx + + client, requests, sleeps, responses = rate_limited_client + responses.extend([ + httpx.Response(429, headers={} if value is None else {"Retry-After": value}), + httpx.Response(200, json={}), + ]) + client.open_campaign({}) + assert sleeps == [expected] + assert len(requests) == 2 + + +@pytest.mark.parametrize("offset,expected", [(3, 3.0), (-10, 0.0), (60, 5.0)]) +def test_retry_after_http_date(rate_limited_client, monkeypatch, offset, expected): + import httpx + from datetime import datetime, timedelta, timezone + from email.utils import format_datetime + + now = datetime(2026, 11, 11, 12, 0, tzinfo=timezone.utc) + + class Clock(datetime): + @classmethod + def now(cls, tz=None): + return now + + monkeypatch.setattr(hosted_sync, "datetime", Clock) + client, _, sleeps, responses = rate_limited_client + responses.extend([ + httpx.Response(429, headers={"Retry-After": format_datetime(now + timedelta(seconds=offset), usegmt=True)}), + httpx.Response(200, json={}), + ]) + client.open_campaign({}) + assert sleeps == [expected] + + +def test_repeated_429_exhausts_retries_and_keeps_pending_bundle(rate_limited_client, tmp_path): + import httpx + + client, requests, sleeps, responses = rate_limited_client + responses.extend(httpx.Response( + 429, headers={"Retry-After": "999"}, + json={"error": {"code": "rate_limit_exceeded", "message": "wait"}}, + ) for _ in range(3)) + outcome = sync_local_campaign( + _bundle(tmp_path, campaign_id="limited", secret_in_trace=True), + api_url=client.api_url, client=client, + ) + assert not outcome.ok + assert outcome.http_status == 429 + assert outcome.error_code == "rate_limit_exceeded" + assert len(requests) == 3 + assert sleeps == [5.0, 5.0] + assert outcome.pending_path.is_file() + saved = outcome.pending_path.read_text(encoding="utf-8") + assert SECRET_VALUE not in saved + assert FAKE_TOKEN not in saved + assert json.loads(saved)["payloads"]["batch"]["redaction"]["applied"] is True + + +@pytest.mark.parametrize("status", [400, 401, 403, 409, 413, 500]) +def test_non_429_errors_do_not_retry(rate_limited_client, status): + import httpx + + client, requests, sleeps, responses = rate_limited_client + responses.append(httpx.Response(status, json={"error": {"code": "original_error"}})) + with pytest.raises(HostedSyncError) as caught: + client.open_campaign({}) + assert caught.value.http_status == status + assert caught.value.error_code == "original_error" + assert len(requests) == 1 + assert sleeps == [] + + +def test_retry_stops_on_a_different_error(rate_limited_client): + import httpx + + client, requests, sleeps, responses = rate_limited_client + responses.extend([httpx.Response(429, headers={"Retry-After": "1"}), httpx.Response(401, json={})]) + with pytest.raises(HostedSyncError) as caught: + client.open_campaign({}) + assert caught.value.http_status == 401 + assert len(requests) == 2 + assert sleeps == [1.0] + + +def test_transport_errors_are_not_retried(rate_limited_client): + import httpx + + client, requests, sleeps, responses = rate_limited_client + responses.append(httpx.ReadTimeout("offline timeout")) + with pytest.raises(httpx.ReadTimeout): + client.open_campaign({}) + assert len(requests) == 1 + assert sleeps == [] + + +def test_sync_recovers_from_429_without_regenerating_or_unredacting_payloads(rate_limited_client, tmp_path): + import httpx + + client, requests, sleeps, responses = rate_limited_client + for _ in range(4): + responses.extend([httpx.Response(429, headers={"Retry-After": "0"}), httpx.Response(200, json={})]) + outcome = sync_local_campaign( + _bundle(tmp_path, violated=True, secret_in_trace=True, with_regression=True), + api_url=client.api_url, client=client, + ) + assert outcome.ok + assert sleeps == [0.0] * 4 + assert len(requests) == 8 + for index in range(0, len(requests), 2): + assert requests[index].content == requests[index + 1].content + assert SECRET_VALUE.encode() not in requests[index].content + assert json.loads(requests[index].content)["redaction"]["applied"] is True + + +def test_retry_support_does_not_bypass_disabled_redaction(rate_limited_client, tmp_path, monkeypatch): + client, requests, sleeps, _ = rate_limited_client + monkeypatch.setenv("MUTINY_DISABLE_SECRET_REDACTION", "1") + outcome = sync_local_campaign(_bundle(tmp_path), api_url=client.api_url, client=client) + assert not outcome.ok + assert outcome.error_code == "redaction_disabled" + assert requests == sleeps == []