From d137cd891774c87860c5a7aa6a20728b7374dc4c Mon Sep 17 00:00:00 2001 From: lukeeexd <31347888+lukeeexd@users.noreply.github.com> Date: Mon, 14 Sep 2026 20:32:26 +0100 Subject: [PATCH 1/4] v0.18.2-0036: fix #1009 planned commit partial-failure reporting and Dispatcharr 429 backoff POST /api/channel-pipeline/run/commit returned 502 with an empty completed_writes list whenever the first replayed write failed, which a caller could not tell apart from a gateway failure. The trigger was Dispatcharr rate limiting: no code path in the client handled 429 and every write called raise_for_status, so one 429 aborted the fail-fast replay while the collect-and-continue /run path shrugged it off. - dispatcharr_client: retry 429 in _request and _login with bounded exponential backoff (1s, 2s, 4s; cap 10s), honouring Retry-After; a spent budget surfaces as an HTTPStatusError carrying the 429. - pipeline_write_plan: PartialReplayError carries failed_write, not_applied and pre_mutation. pre_mutation is True only when nothing completed, compensation was clean, and the first write was provably rejected (4xx or connection refused) before upstream mutated. - routers/channel_pipeline: a partial replay returns 424 Failed Dependency with execution_id, failed_index, failed_write, pre_mutation, completed_writes, not_applied and compensation_errors. The commit stays synchronous; the 30s request-timeout exposure raised in #1009 is left for a separate maintainer decision. - tests: replay classification (429 first / 429 after a landed write / timeout), client retry and budget behaviour, router 424 contract. Closes #1009 Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 1 + backend/dispatcharr_client.py | 115 ++++++++++++---- backend/routers/channel_pipeline.py | 16 ++- backend/services/pipeline_write_plan.py | 52 +++++++- .../tests/routers/test_channel_pipeline.py | 66 +++++++++ .../tests/services/test_event_sync_cleanup.py | 2 +- .../services/test_pipeline_write_plan.py | 65 ++++++++- .../test_dispatcharr_client_rate_limit.py | 126 ++++++++++++++++++ docs/user_guide/integrations/mcp.md | 16 ++- 9 files changed, 422 insertions(+), 37 deletions(-) create mode 100644 backend/tests/unit/test_dispatcharr_client_rate_limit.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 990be4275..cae98c24c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Fixed - **Event Sync pre-flight names the real constraint when the master group is not M3U-backed ([#1007](https://github.com/MotWakorb/enhancedchannelmanager/issues/1007); build 0.18.2-0036).** The `group_settings_found` failure for a master group used to read as a lookup error ("was not found in the M3U account's group settings — the group may have been removed or renamed"), which sent operators looking for a missing group when the group existed but was hand-curated. The message now states that the master must come from an M3U account with `auto_channel_sync` ON because Dispatcharr owns master-channel lifecycle, that a hand-curated channel group is not supported as a master, and only then mentions the removed/renamed case. The check id, failure shape and behaviour are unchanged; `docs/event_sync.md` gains the same note under "Pick the master group" and "Pre-flight checks". The diagnosis is scoped to what was actually checked: a provider-scoped master reports the missing provider/group association (and whether another account carries the group) instead of claiming no account carries it, and "M3U-backed" includes a whole-group Channel Group Override target of an auto-synced group. +- **Planned pipeline commit reports a partial failure honestly and survives Dispatcharr rate limiting (GitHub #1009; build 0.18.2-0036).** `POST /api/channel-pipeline/run/commit` returned `502` with an empty `completed_writes` list whenever the first replayed write failed, which a caller could not tell apart from a gateway failure. A partial replay now returns `424 Failed Dependency` with the failed write, the writes applied and not applied, compensation errors, the execution id, and a `pre_mutation` flag that is true only when nothing landed and the first write was provably rejected before upstream mutated. The Dispatcharr client retries `429 Too Many Requests` on API requests and login with bounded backoff, honouring `Retry-After`, and surfaces a spent budget as an HTTP status error carrying the 429; previously no code path handled a 429 and a single one aborted the replay. The commit endpoint remains synchronous; the 30s request-timeout exposure raised in #1009 is left for a separate decision. - **Section navigation highlights the settled reading position (bead `enhancedchannelmanager-2896r.19`; build 0.18.2-0035).** Scheduled Tasks keeps the chosen section highlighted after scrolling in either direction, including bottom-clamped targets and reloaded links. Refresh preserves the reader's position without replaying a consumed section link. diff --git a/backend/dispatcharr_client.py b/backend/dispatcharr_client.py index a9c972c10..505a86139 100644 --- a/backend/dispatcharr_client.py +++ b/backend/dispatcharr_client.py @@ -253,6 +253,44 @@ def dispatcharr_version_advisory(version) -> Optional[str]: ) +# 429 handling (GH #1009). Dispatcharr rate-limits its JWT endpoints (login is +# 3/min per IP) and, under bulk writes, ordinary API calls too. Before this no +# code path handled 429: every write called ``raise_for_status`` and one +# rate-limited response aborted a planned pipeline replay. A 429 is retried +# with bounded exponential backoff, honouring ``Retry-After`` when present. +RATE_LIMIT_MAX_RETRIES = 3 +RATE_LIMIT_BACKOFF_BASE = 1.0 +RATE_LIMIT_BACKOFF_CAP = 10.0 +# Indirection so tests can patch the sleeper without touching asyncio itself. +_sleep = asyncio.sleep + + +def _rate_limited_error(response: httpx.Response) -> httpx.HTTPStatusError: + """Dispatcharr kept answering 429 after the retry budget was spent. + + A plain ``HTTPStatusError`` (no subclass: the contract sweep forbids + classes in this module inheriting from outside it) whose ``response`` + carries the 429, so callers can tell a rate-limit rejection, which + upstream never applied, from every other failure by status code. + """ + return httpx.HTTPStatusError( + f"Dispatcharr rate limited (429) after {RATE_LIMIT_MAX_RETRIES} retries", + request=getattr(response, "request", None), + response=response, + ) + + +def _rate_limit_delay(response: httpx.Response, attempt: int) -> float: + """Seconds to wait before retry ``attempt`` (0-based) of a 429.""" + retry_after = response.headers.get("Retry-After") if response.headers is not None else None + if retry_after: + try: + return max(0.0, float(retry_after)) + except ValueError: + pass # HTTP-date form or garbage: fall back to backoff + return min(RATE_LIMIT_BACKOFF_BASE * (2 ** attempt), RATE_LIMIT_BACKOFF_CAP) + + class DispatcharrClient: """API client for Dispatcharr with JWT authentication.""" @@ -305,13 +343,24 @@ async def _login(self) -> None: """Authenticate and obtain JWT tokens.""" logger.debug("[DISPATCHARR] Authenticating to Dispatcharr at %s", self.base_url) try: - response = await self._client.post( - f"{self.base_url}/api/accounts/token/", - json={ - "username": self.settings.username, - "password": self.settings.password, - }, - ) + for attempt in range(RATE_LIMIT_MAX_RETRIES + 1): + response = await self._client.post( + f"{self.base_url}/api/accounts/token/", + json={ + "username": self.settings.username, + "password": self.settings.password, + }, + ) + if response.status_code != 429: + break + if attempt >= RATE_LIMIT_MAX_RETRIES: + raise _rate_limited_error(response) + delay = _rate_limit_delay(response, attempt) + logger.warning( + "[DISPATCHARR] Login rate limited (429); retrying in %.1fs (attempt %d/%d)", + delay, attempt + 1, RATE_LIMIT_MAX_RETRIES, + ) + await _sleep(delay) response.raise_for_status() data = response.json() self.access_token = data["access"] @@ -407,22 +456,7 @@ async def _request( logger.debug("[DISPATCHARR] Using extended timeout (%ss) for EPG grid request", request_timeout) try: - response = await self._client.request( - method, - f"{self.base_url}{path}", - headers=headers, - timeout=request_timeout, - **kwargs, - ) - - # If unauthorized in JWT mode, try refreshing token and retry. - # In api-key mode a 401 is terminal (the key is invalid or revoked), - # and callers that opted out of the retry take the 401 as terminal - # too rather than risk a rate-limited re-login (see the docstring). - if response.status_code == 401 and not self._uses_api_key and retry_on_401: - logger.debug("[DISPATCHARR] Got 401, refreshing token and retrying: %s", method) - await self._refresh_access_token() - headers["Authorization"] = f"Bearer {self.access_token}" + for attempt in range(RATE_LIMIT_MAX_RETRIES + 1): response = await self._client.request( method, f"{self.base_url}{path}", @@ -431,6 +465,41 @@ async def _request( **kwargs, ) + # If unauthorized in JWT mode, try refreshing token and retry. + # In api-key mode a 401 is terminal (the key is invalid or revoked), + # and callers that opted out of the retry take the 401 as terminal + # too rather than risk a rate-limited re-login (see the docstring). + if response.status_code == 401 and not self._uses_api_key and retry_on_401: + logger.debug("[DISPATCHARR] Got 401, refreshing token and retrying: %s", method) + await self._refresh_access_token() + headers["Authorization"] = f"Bearer {self.access_token}" + response = await self._client.request( + method, + f"{self.base_url}{path}", + headers=headers, + timeout=request_timeout, + **kwargs, + ) + + # Rate limited: back off and retry (GH #1009). Exhausting the + # budget raises an HTTPStatusError carrying the 429 so callers + # can tell "upstream rejected this before applying it" from + # other failures. + if response.status_code != 429: + break + if attempt >= RATE_LIMIT_MAX_RETRIES: + logger.warning( + "[DISPATCHARR] Rate limited (429) on %s after %d retries; giving up", + method, RATE_LIMIT_MAX_RETRIES, + ) + raise _rate_limited_error(response) + delay = _rate_limit_delay(response, attempt) + logger.warning( + "[DISPATCHARR] Rate limited (429) on %s; retrying in %.1fs (attempt %d/%d)", + method, delay, attempt + 1, RATE_LIMIT_MAX_RETRIES, + ) + await _sleep(delay) + if response.status_code >= 400: logger.warning("[DISPATCHARR] API request failed: %s - status: %s", method, response.status_code) else: diff --git a/backend/routers/channel_pipeline.py b/backend/routers/channel_pipeline.py index d7313f9a3..f3e5b6a8e 100644 --- a/backend/routers/channel_pipeline.py +++ b/backend/routers/channel_pipeline.py @@ -2025,15 +2025,29 @@ async def commit_auto_creation_pipeline(request: CommitPipelinePlanRequest, _adm except PartialReplayError as exc: partial_replay = { "failed_index": exc.failed_index, + "failed_write": exc.failed_write, "completed_targets": exc.completed, + "not_applied": exc.not_applied, + "pre_mutation": exc.pre_mutation, "compensation_errors": exc.compensation_errors, } _mark_execution_failed(execution_id, exc, partial_replay=partial_replay) + # 424 Failed Dependency, not 502 (GH #1009): the failure is ECM's + # own, recorded on the execution row, and the body says exactly + # what was and was not applied. A 502/504 is what proxies emit and + # would leave the caller unable to tell whether a retry is safe. + # ``pre_mutation`` is True only when nothing landed AND the first + # write was provably rejected (e.g. 429) before upstream mutated. raise HTTPException( - status_code=502, + status_code=424, detail={ "message": "pipeline replay partially failed", + "execution_id": execution_id, + "failed_index": exc.failed_index, + "failed_write": exc.failed_write, + "pre_mutation": exc.pre_mutation, "completed_writes": exc.completed, + "not_applied": exc.not_applied, "compensation_errors": exc.compensation_errors, }, ) from exc diff --git a/backend/services/pipeline_write_plan.py b/backend/services/pipeline_write_plan.py index b9d17b602..145bcb408 100644 --- a/backend/services/pipeline_write_plan.py +++ b/backend/services/pipeline_write_plan.py @@ -5,6 +5,8 @@ from dataclasses import dataclass, field from typing import Any +import httpx + from services.mutation_plan_store import canonical_hash @@ -85,13 +87,44 @@ def accounting(self) -> dict[str, int]: class PartialReplayError(RuntimeError): - """Upstream has no transaction; exposes exactly how far replay reached.""" + """Upstream has no transaction; exposes exactly how far replay reached. + + ``completed`` and ``not_applied`` are target strings (``method:first_arg``) + so a caller can see what landed and what did not. ``pre_mutation`` is + True only when NO write completed AND the failing write's exception proves + upstream rejected it before mutating anything (GH #1009); it is the one + case in which a fresh prepare + commit is known to be safe. + """ - def __init__(self, failed_index: int, completed: list[str], compensation_errors: list[str]): + def __init__( + self, failed_index: int, completed: list[str], compensation_errors: list[str], + *, failed_write: str = "", not_applied: list[str] | None = None, + pre_mutation: bool = False, + ): super().__init__(f"pipeline replay failed at write {failed_index}") self.failed_index = failed_index self.completed = completed self.compensation_errors = compensation_errors + self.failed_write = failed_write + self.not_applied = list(not_applied or []) + self.pre_mutation = pre_mutation + + +def _write_target(write: PlannedWrite, args: list[Any]) -> str: + return f"{write.method}:{args[0] if args else ''}" + + +def _failure_precludes_mutation(exc: BaseException) -> bool: + """True only when the failure provably happened before upstream mutated. + + A 4xx response is a rejection (429 rate limit, 400 validation, 404 gone) + and a refused connection never reached Dispatcharr. Timeouts, dropped + connections, 5xx, and every unknown exception are treated as "may have + landed" so a caller is never told a retry is safe when it is not. + """ + if isinstance(exc, httpx.HTTPStatusError): + return 400 <= exc.response.status_code < 500 + return isinstance(exc, httpx.ConnectError) class PlanningDispatcharrClient: @@ -313,12 +346,17 @@ def mapped(value: Any) -> Any: }) except Exception as compensation_exc: # noqa: BLE001 compensation_errors.append(f"{done.method}: {compensation_exc}") - completed_targets = [ - f"{item[0].method}:{item[1][0] if item[1] else ''}" - for item in completed - ] + completed_targets = [_write_target(item[0], item[1]) for item in completed] + failed_index = len(completed) + not_applied = [_write_target(write, write.args) for write in plan.writes[failed_index:]] raise PartialReplayError( - len(completed), completed_targets, compensation_errors + failed_index, completed_targets, compensation_errors, + failed_write=not_applied[0] if not_applied else "", + not_applied=not_applied, + pre_mutation=( + not completed and not compensation_errors + and _failure_precludes_mutation(exc) + ), ) from exc return results, remap diff --git a/backend/tests/routers/test_channel_pipeline.py b/backend/tests/routers/test_channel_pipeline.py index ecf6e7a40..a303b5071 100644 --- a/backend/tests/routers/test_channel_pipeline.py +++ b/backend/tests/routers/test_channel_pipeline.py @@ -5394,3 +5394,69 @@ async def test_import_round_trips_flag(self, async_client, test_session): by_name = {r["name"]: r for r in rules} assert by_name["Imported Folded Rule"]["fold_match_key"] is True assert by_name["Imported Legacy Rule"]["fold_match_key"] is False + + +class TestCommitPartialFailureGH1009: + """A partial planned-run replay is an application-level failure, not a 502. + + The handler is invoked directly (as ``test_event_sync_cleanup`` does) so + the assertion is on the router's own contract, independent of middleware. + """ + + @pytest.mark.asyncio + async def test_partial_replay_returns_424_with_applied_and_not_applied(self, test_engine): + from fastapi import HTTPException + from sqlalchemy.orm import sessionmaker + from routers import channel_pipeline as router + from services import mutation_plan_store as store + from services.mutation_plan_store import canonical_hash + from services.pipeline_write_plan import PartialReplayError + + payload = { + "request": {"m3u_account_ids": None, "rule_ids": [7]}, + "result": {"event_sync": [], "planned_review_candidates": [], "execution_log": []}, + "write_plan": { + "writes": [ + {"method": "update_channel", "args": [7, {"name": "A"}], "kwargs": {}, "event_sync": None}, + {"method": "delete_channel", "args": [8], "kwargs": {}, "event_sync": None}, + ], + "channel_preconditions": {}, "group_preconditions": {}, "profile_preconditions": {}, + }, + "snapshot": [], + } + fresh_store = store.MutationPlanStore() + plan = fresh_store.create( + "channel_pipeline", payload, canonical_hash(router._canonical_pipeline_decision(payload)), + ) + failure = PartialReplayError( + 0, [], [], failed_write="update_channel:7", + not_applied=["update_channel:7", "delete_channel:8"], pre_mutation=True, + ) + marked: dict = {} + + def capture_failed(execution_id, error, *, partial_replay=None): + marked["execution_id"] = execution_id + marked["partial_replay"] = partial_replay + + with patch.object(store, "mutation_plan_store", fresh_store), patch.object(router, "_ensure_engine", AsyncMock(return_value=MagicMock(client=object()))), patch.object(router, "_compute_pipeline_plan_payload", AsyncMock(return_value=payload)), patch("services.pipeline_write_plan.validate_read_set", AsyncMock()), patch("services.pipeline_write_plan.replay_write_plan", AsyncMock(side_effect=failure)), patch.object(router, "_mark_execution_failed", side_effect=capture_failed), patch.object(router, "get_session", sessionmaker(bind=test_engine)): + with pytest.raises(HTTPException) as error: + await router.commit_auto_creation_pipeline( + router.CommitPipelinePlanRequest( + plan_id=plan.plan_id, plan_hash=plan.payload_hash, phase="execute", + ), + _admin=None, + ) + + assert error.value.status_code == 424 + detail = error.value.detail + assert detail["message"] == "pipeline replay partially failed" + assert isinstance(detail["execution_id"], int) + assert detail["failed_index"] == 0 + assert detail["failed_write"] == "update_channel:7" + assert detail["pre_mutation"] is True + assert detail["completed_writes"] == [] + assert detail["not_applied"] == ["update_channel:7", "delete_channel:8"] + assert detail["compensation_errors"] == [] + assert marked["execution_id"] == detail["execution_id"] + assert marked["partial_replay"]["failed_write"] == "update_channel:7" + assert marked["partial_replay"]["pre_mutation"] is True diff --git a/backend/tests/services/test_event_sync_cleanup.py b/backend/tests/services/test_event_sync_cleanup.py index 1726d2e65..5e8b30ee7 100644 --- a/backend/tests/services/test_event_sync_cleanup.py +++ b/backend/tests/services/test_event_sync_cleanup.py @@ -441,7 +441,7 @@ async def compute(request): with pytest.raises(HTTPException) as error: await router.commit_auto_creation_pipeline(router.CommitPipelinePlanRequest( plan_id=prepared["plan_id"], plan_hash=prepared["plan_hash"], phase="execute"), _admin=None) - assert error.value.status_code == 502 + assert error.value.status_code == 424 assert "uncertain_cleanup_outcome" in error.value.detail["compensation_errors"] assert upstream.channel["streams"] == [1] with Session(db) as session: diff --git a/backend/tests/services/test_pipeline_write_plan.py b/backend/tests/services/test_pipeline_write_plan.py index b3492451a..74128da11 100644 --- a/backend/tests/services/test_pipeline_write_plan.py +++ b/backend/tests/services/test_pipeline_write_plan.py @@ -2,11 +2,12 @@ from pathlib import Path from unittest.mock import AsyncMock +import httpx import pytest from services.pipeline_write_plan import ( PIPELINE_INTERNAL_SIDE_EFFECTS, PIPELINE_WRITE_METHODS, PlanningDispatcharrClient, PipelineWritePlan, - PlannedWrite, replay_write_plan, + PartialReplayError, PlannedWrite, replay_write_plan, ) @@ -104,3 +105,65 @@ async def test_drift_rejects_before_any_replay_write(): with pytest.raises(ValueError, match="drifted"): await replay_write_plan(live, plan) live.delete_channel.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# GH #1009: a partial replay must say which write failed, what was and was +# not applied, and whether the failure provably happened before any mutation. +# --------------------------------------------------------------------------- + + +def _rate_limited() -> httpx.HTTPStatusError: + response = httpx.Response(429, request=httpx.Request("PATCH", "http://dispatcharr/x")) + return httpx.HTTPStatusError("429", request=response.request, response=response) + + +def _two_write_plan() -> PipelineWritePlan: + return PipelineWritePlan( + writes=[ + PlannedWrite("update_channel", [7, {"name": "A"}], {}), + PlannedWrite("delete_channel", [8], {}), + ], + ) + + +@pytest.mark.asyncio +async def test_first_write_rejected_with_429_is_reported_as_pre_mutation(): + live = AsyncMock() + live.update_channel.side_effect = _rate_limited() + with pytest.raises(PartialReplayError) as error: + await replay_write_plan(live, _two_write_plan()) + exc = error.value + assert exc.failed_index == 0 + assert exc.failed_write == "update_channel:7" + assert exc.completed == [] + assert exc.not_applied == ["update_channel:7", "delete_channel:8"] + assert exc.pre_mutation is True + live.delete_channel.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_failure_after_a_landed_write_is_not_pre_mutation(): + live = AsyncMock() + live.update_channel.return_value = {"id": 7} + live.delete_channel.side_effect = _rate_limited() + with pytest.raises(PartialReplayError) as error: + await replay_write_plan(live, _two_write_plan()) + exc = error.value + assert exc.failed_index == 1 + assert exc.failed_write == "delete_channel:8" + assert exc.completed == ["update_channel:7"] + assert exc.not_applied == ["delete_channel:8"] + assert exc.pre_mutation is False + + +@pytest.mark.asyncio +async def test_first_write_timeout_is_not_claimed_pre_mutation(): + """A lost response may have landed upstream; never claim retry is safe.""" + live = AsyncMock() + live.update_channel.side_effect = httpx.ReadTimeout("slow", request=httpx.Request("PATCH", "http://d/x")) + with pytest.raises(PartialReplayError) as error: + await replay_write_plan(live, _two_write_plan()) + assert error.value.failed_index == 0 + assert error.value.completed == [] + assert error.value.pre_mutation is False diff --git a/backend/tests/unit/test_dispatcharr_client_rate_limit.py b/backend/tests/unit/test_dispatcharr_client_rate_limit.py new file mode 100644 index 000000000..d8497e76f --- /dev/null +++ b/backend/tests/unit/test_dispatcharr_client_rate_limit.py @@ -0,0 +1,126 @@ +"""Dispatcharr 429 handling in ``DispatcharrClient`` (GH #1009). + +Before this, no code path in the client handled ``429 Too Many Requests``: +every write called ``raise_for_status`` and a single rate-limited response +aborted a planned pipeline replay. The client now retries a 429 with +bounded backoff, honouring ``Retry-After`` when Dispatcharr sends one, and +raises an ``HTTPStatusError`` carrying the 429 once the budget is spent so +callers can tell a rate-limit rejection from every other failure. +""" +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx + +import dispatcharr_client +from config import DispatcharrSettings +from dispatcharr_client import DispatcharrClient + + +def _api_key_client() -> DispatcharrClient: + return DispatcharrClient(DispatcharrSettings( + url="http://dispatcharr:8000", auth_method="api_key", dispatcharr_api_key="k", + )) + + +def _jwt_client() -> DispatcharrClient: + return DispatcharrClient(DispatcharrSettings( + url="http://dispatcharr:8000", auth_method="password", username="u", password="p", + )) + + +def _response(status_code: int, headers: dict | None = None, json_body=None) -> httpx.Response: + return httpx.Response( + status_code, headers=headers or {}, json=json_body if json_body is not None else {}, + request=httpx.Request("GET", "http://dispatcharr:8000/api/x/"), + ) + + +@pytest.mark.asyncio +async def test_request_retries_429_with_exponential_backoff_then_succeeds(): + client = _api_key_client() + sleeps: list[float] = [] + try: + client._client.request = AsyncMock(side_effect=[ + _response(429), _response(429), _response(200, json_body={"ok": True}), + ]) + with patch.object(dispatcharr_client, "_sleep", AsyncMock(side_effect=sleeps.append)): + response = await client._request("GET", "/api/x/") + finally: + await client._client.aclose() + assert response.status_code == 200 + assert client._client.request.await_count == 3 + assert sleeps == [1.0, 2.0] + + +@pytest.mark.asyncio +async def test_request_honours_retry_after_header(): + client = _api_key_client() + sleeps: list[float] = [] + try: + client._client.request = AsyncMock(side_effect=[ + _response(429, headers={"Retry-After": "7"}), _response(200), + ]) + with patch.object(dispatcharr_client, "_sleep", AsyncMock(side_effect=sleeps.append)): + await client._request("GET", "/api/x/") + finally: + await client._client.aclose() + assert sleeps == [7.0] + + +@pytest.mark.asyncio +async def test_request_raises_429_status_error_once_retry_budget_is_spent(): + client = _api_key_client() + try: + client._client.request = AsyncMock(return_value=_response(429)) + with patch.object(dispatcharr_client, "_sleep", AsyncMock()): + with pytest.raises(httpx.HTTPStatusError) as error: + await client._request("GET", "/api/x/") + finally: + await client._client.aclose() + assert error.value.response.status_code == 429 + assert "rate limited" in str(error.value) + assert client._client.request.await_count == dispatcharr_client.RATE_LIMIT_MAX_RETRIES + 1 + + +@pytest.mark.asyncio +async def test_login_retries_429_then_stores_tokens(): + client = _jwt_client() + sleeps: list[float] = [] + try: + client._client.post = AsyncMock(side_effect=[ + _response(429), _response(200, json_body={"access": "A", "refresh": "R"}), + ]) + with patch.object(dispatcharr_client, "_sleep", AsyncMock(side_effect=sleeps.append)): + await client._login() + finally: + await client._client.aclose() + assert (client.access_token, client.refresh_token) == ("A", "R") + assert sleeps == [1.0] + + +@pytest.mark.asyncio +async def test_login_raises_429_status_error_once_retry_budget_is_spent(): + client = _jwt_client() + try: + client._client.post = AsyncMock(return_value=_response(429)) + with patch.object(dispatcharr_client, "_sleep", AsyncMock()): + with pytest.raises(httpx.HTTPStatusError) as error: + await client._login() + finally: + await client._client.aclose() + assert error.value.response.status_code == 429 + assert client.access_token is None + + +@pytest.mark.asyncio +async def test_login_budget_failure_leaves_no_token(): + client = _jwt_client() + try: + client._client.post = AsyncMock(return_value=_response(429)) + with patch.object(dispatcharr_client, "_sleep", AsyncMock()): + with pytest.raises(httpx.HTTPStatusError): + await client._login() + finally: + await client._client.aclose() + assert client.access_token is None diff --git a/docs/user_guide/integrations/mcp.md b/docs/user_guide/integrations/mcp.md index ab70d26e7..222dd7769 100644 --- a/docs/user_guide/integrations/mcp.md +++ b/docs/user_guide/integrations/mcp.md @@ -740,7 +740,15 @@ the planned-run lock before the first write. Any difference in its canonical decision or exact write payload requires a new preview. ECM persists the exact execution program and target-scoped rollback snapshot immediately before replay, then compensates reversible writes in reverse order if a later write fails. A -failure response lists target-specific completed writes and any compensation -failures. Deletes and channel-profile membership changes cannot always be -restored with the same upstream identifiers; inspect the execution record and -rollback snapshot before retrying a partially failed commit. +partial failure returns `424 Failed Dependency` (not a proxy-style 502) whose +body names the write that failed, lists target-specific completed writes and +the writes that were not applied, reports any compensation failures, and +carries the execution id to inspect. Its `pre_mutation` flag is `true` only +when nothing was applied and Dispatcharr provably rejected the first write +before mutating anything (for example a rate-limit rejection); that is the one +case in which a fresh prepare and commit is known to be safe. Deletes and +channel-profile membership changes cannot always be restored with the same +upstream identifiers; inspect the execution record and rollback snapshot before +retrying a partially failed commit. Transient `429 Too Many Requests` answers +from Dispatcharr are retried with bounded backoff before they count as a +failure. From 525e47242d0d714710c9b1a0894c1cfbd7ec1eca Mon Sep 17 00:00:00 2001 From: lukeeexd <31347888+lukeeexd@users.noreply.github.com> Date: Fri, 18 Sep 2026 21:43:33 +0100 Subject: [PATCH 2/4] fix(channel-pipeline): address #1010 review on partial-replay reporting and 429 backoff (#1009) 1. Outcome classes are disjoint. The failed write carries failed_outcome "rejected" (4xx or refused connection: upstream provably did not mutate) or "unknown" (timeout, dropped connection, 5xx: it may have landed). not_applied lists only the writes AFTER the failed one, which were never attempted; a landed PATCH with a lost response is no longer reported as not applied. completed_writes is documented as forward-call history, not current upstream state. Exposed on the 424 body, the execution log entry and the snapshot evidence. 2. Descriptors name the resolved upstream id: an update recorded against temp id -1 that PATCHed channel 101 reports update_channel:101; a completed create reports the resource it produced (create_channel#0->101) so a failed compensation can be located; an unresolved future target renders as pending(-2) instead of failing rendering. 3. Descriptors never stringify argument dictionaries or URLs. A create_logo payload with a credentialed URL is identified by plan position (create_logo#0) in failed_write, not_applied, completed and str(exc). 4. Retry-After is validated (finite, non-negative) and every call's 429 waits are bounded by an explicit 30s total budget shared by login and API requests. A server-directed delay beyond the budget raises the 429 HTTPStatusError immediately instead of sleeping (an infinite login sleep previously held _auth_lock); the lock is released on that path. 5. Retry-After in HTTP-date form is parsed to its remaining non-negative interval under the same budget; garbage still falls back to backoff. Tests: lost-response vs confirmed-rejection, resolved ids after a dependent update failure, compensation-failure naming, pending targets, credential canary absent from every diagnostic field, mapping error before the call, Retry-After parser edge cases, over-budget seconds and dates, cumulative budget, invalid values, auth-lock release. Router test updated for failed_outcome and the narrowed not_applied. Docs and changelog aligned. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01MGuX7qVpTp99bzYZXgLJMm --- CHANGELOG.md | 2 +- backend/dispatcharr_client.py | 102 ++++++++-- backend/routers/channel_pipeline.py | 6 + backend/services/pipeline_write_plan.py | 112 +++++++++-- .../tests/routers/test_channel_pipeline.py | 10 +- .../services/test_pipeline_write_plan.py | 158 +++++++++++++++- .../test_dispatcharr_client_rate_limit.py | 179 ++++++++++++++++++ docs/user_guide/integrations/mcp.md | 38 +++- 8 files changed, 563 insertions(+), 44 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cae98c24c..2226b4bc4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Fixed - **Event Sync pre-flight names the real constraint when the master group is not M3U-backed ([#1007](https://github.com/MotWakorb/enhancedchannelmanager/issues/1007); build 0.18.2-0036).** The `group_settings_found` failure for a master group used to read as a lookup error ("was not found in the M3U account's group settings — the group may have been removed or renamed"), which sent operators looking for a missing group when the group existed but was hand-curated. The message now states that the master must come from an M3U account with `auto_channel_sync` ON because Dispatcharr owns master-channel lifecycle, that a hand-curated channel group is not supported as a master, and only then mentions the removed/renamed case. The check id, failure shape and behaviour are unchanged; `docs/event_sync.md` gains the same note under "Pick the master group" and "Pre-flight checks". The diagnosis is scoped to what was actually checked: a provider-scoped master reports the missing provider/group association (and whether another account carries the group) instead of claiming no account carries it, and "M3U-backed" includes a whole-group Channel Group Override target of an auto-synced group. -- **Planned pipeline commit reports a partial failure honestly and survives Dispatcharr rate limiting (GitHub #1009; build 0.18.2-0036).** `POST /api/channel-pipeline/run/commit` returned `502` with an empty `completed_writes` list whenever the first replayed write failed, which a caller could not tell apart from a gateway failure. A partial replay now returns `424 Failed Dependency` with the failed write, the writes applied and not applied, compensation errors, the execution id, and a `pre_mutation` flag that is true only when nothing landed and the first write was provably rejected before upstream mutated. The Dispatcharr client retries `429 Too Many Requests` on API requests and login with bounded backoff, honouring `Retry-After`, and surfaces a spent budget as an HTTP status error carrying the 429; previously no code path handled a 429 and a single one aborted the replay. The commit endpoint remains synchronous; the 30s request-timeout exposure raised in #1009 is left for a separate decision. +- **Planned pipeline commit reports a partial failure honestly and survives Dispatcharr rate limiting (GitHub #1009; build 0.18.2-0036).** `POST /api/channel-pipeline/run/commit` returned `502` with an empty `completed_writes` list whenever the first replayed write failed, which a caller could not tell apart from a gateway failure. A partial replay now returns `424 Failed Dependency` with the execution id, the writes that completed (forward-call history), the failed write with a `failed_outcome` of `rejected` (upstream provably refused it) or `unknown` (a lost response or 5xx that may have landed), the later writes that were never attempted, compensation errors, and a `pre_mutation` flag that is true only when nothing landed and the failed write was rejected. Write descriptors name the resolved Dispatcharr id where known, mark a not-yet-created target as pending, and never include payload contents such as logo URLs. The Dispatcharr client retries `429 Too Many Requests` on API requests and login with bounded exponential backoff, honouring `Retry-After` in both its seconds and HTTP-date forms, within an explicit 30-second total wait budget per call; a `Retry-After` beyond the budget, or a spent retry budget, surfaces as an HTTP status error carrying the 429 instead of an unbounded wait. Previously no code path handled a 429 and a single one aborted the replay. The commit endpoint remains synchronous; the 30s request-timeout exposure raised in #1009 is left for a separate decision. - **Section navigation highlights the settled reading position (bead `enhancedchannelmanager-2896r.19`; build 0.18.2-0035).** Scheduled Tasks keeps the chosen section highlighted after scrolling in either direction, including bottom-clamped targets and reloaded links. Refresh preserves the reader's position without replaying a consumed section link. diff --git a/backend/dispatcharr_client.py b/backend/dispatcharr_client.py index 505a86139..a3b162fb9 100644 --- a/backend/dispatcharr_client.py +++ b/backend/dispatcharr_client.py @@ -2,8 +2,12 @@ import hashlib import hmac import json +import math import re import secrets +import time +from datetime import timezone +from email.utils import parsedate_to_datetime import httpx import logging from typing import Optional @@ -261,11 +265,19 @@ def dispatcharr_version_advisory(version) -> Optional[str]: RATE_LIMIT_MAX_RETRIES = 3 RATE_LIMIT_BACKOFF_BASE = 1.0 RATE_LIMIT_BACKOFF_CAP = 10.0 -# Indirection so tests can patch the sleeper without touching asyncio itself. +# Explicit finite budget for the TOTAL time one call may spend waiting on 429s +# across all its retries (PR #1010 review item 4). A server-directed +# ``Retry-After`` that would push the cumulative wait past this is surfaced as +# throttling immediately rather than slept on: the login path holds +# ``_auth_lock`` while it waits, so an unbounded wait would stall every +# request behind it. +RATE_LIMIT_MAX_TOTAL_WAIT = 30.0 +# Indirection so tests can patch the sleeper/clock without touching the stdlib. _sleep = asyncio.sleep +_now = time.time -def _rate_limited_error(response: httpx.Response) -> httpx.HTTPStatusError: +def _rate_limited_error(response: httpx.Response, reason: str | None = None) -> httpx.HTTPStatusError: """Dispatcharr kept answering 429 after the retry budget was spent. A plain ``HTTPStatusError`` (no subclass: the contract sweep forbids @@ -273,22 +285,64 @@ def _rate_limited_error(response: httpx.Response) -> httpx.HTTPStatusError: carries the 429, so callers can tell a rate-limit rejection, which upstream never applied, from every other failure by status code. """ + detail = reason or f"after {RATE_LIMIT_MAX_RETRIES} retries" return httpx.HTTPStatusError( - f"Dispatcharr rate limited (429) after {RATE_LIMIT_MAX_RETRIES} retries", + f"Dispatcharr rate limited (429) {detail}", request=getattr(response, "request", None), response=response, ) -def _rate_limit_delay(response: httpx.Response, attempt: int) -> float: - """Seconds to wait before retry ``attempt`` (0-based) of a 429.""" - retry_after = response.headers.get("Retry-After") if response.headers is not None else None - if retry_after: - try: - return max(0.0, float(retry_after)) - except ValueError: - pass # HTTP-date form or garbage: fall back to backoff - return min(RATE_LIMIT_BACKOFF_BASE * (2 ** attempt), RATE_LIMIT_BACKOFF_CAP) +def _parse_retry_after(value: str | None) -> float | None: + """Seconds a ``Retry-After`` header asks us to wait, or None if unusable. + + Accepts both representations RFC 9110 allows: delay-seconds and an + HTTP-date (PR #1010 review item 5). Only finite, non-negative results are + honoured; ``inf``, ``nan``, negatives and garbage yield None so the caller + falls back to exponential backoff instead of sleeping forever or zero. + """ + if not value: + return None + text = value.strip() + try: + seconds = float(text) + except ValueError: + seconds = None + if seconds is not None: + # delay-seconds: only a finite, non-negative number is a valid delay. + if not math.isfinite(seconds) or seconds < 0: + return None + return seconds + try: + when = parsedate_to_datetime(text) + except (TypeError, ValueError, IndexError, OverflowError): + return None + if when is None: + return None + if when.tzinfo is None: + when = when.replace(tzinfo=timezone.utc) + remaining = when.timestamp() - _now() + if not math.isfinite(remaining): + return None + # HTTP-date: a date already in the past means "retry now". + return max(0.0, remaining) + + +def _rate_limit_delay(response: httpx.Response, attempt: int, waited: float) -> float | None: + """Seconds to wait before retry ``attempt`` (0-based) of a 429, or None. + + ``waited`` is the time this call has already spent waiting on 429s. None + means the retry-wait budget (:data:`RATE_LIMIT_MAX_TOTAL_WAIT`) does not + admit another wait: the caller must surface the throttling instead of + retrying early (which the server forbade) or waiting past the budget. + """ + headers = response.headers if response.headers is not None else {} + delay = _parse_retry_after(headers.get("Retry-After")) + if delay is None: + delay = min(RATE_LIMIT_BACKOFF_BASE * (2 ** attempt), RATE_LIMIT_BACKOFF_CAP) + if waited + delay > RATE_LIMIT_MAX_TOTAL_WAIT: + return None + return delay class DispatcharrClient: @@ -343,6 +397,7 @@ async def _login(self) -> None: """Authenticate and obtain JWT tokens.""" logger.debug("[DISPATCHARR] Authenticating to Dispatcharr at %s", self.base_url) try: + waited = 0.0 for attempt in range(RATE_LIMIT_MAX_RETRIES + 1): response = await self._client.post( f"{self.base_url}/api/accounts/token/", @@ -355,11 +410,20 @@ async def _login(self) -> None: break if attempt >= RATE_LIMIT_MAX_RETRIES: raise _rate_limited_error(response) - delay = _rate_limit_delay(response, attempt) + delay = _rate_limit_delay(response, attempt, waited) + if delay is None: + logger.warning( + "[DISPATCHARR] Login rate limited (429); Retry-After exceeds the %.0fs wait budget, giving up", + RATE_LIMIT_MAX_TOTAL_WAIT, + ) + raise _rate_limited_error( + response, f"Retry-After exceeds the {RATE_LIMIT_MAX_TOTAL_WAIT:.0f}s wait budget", + ) logger.warning( "[DISPATCHARR] Login rate limited (429); retrying in %.1fs (attempt %d/%d)", delay, attempt + 1, RATE_LIMIT_MAX_RETRIES, ) + waited += delay await _sleep(delay) response.raise_for_status() data = response.json() @@ -456,6 +520,7 @@ async def _request( logger.debug("[DISPATCHARR] Using extended timeout (%ss) for EPG grid request", request_timeout) try: + waited = 0.0 for attempt in range(RATE_LIMIT_MAX_RETRIES + 1): response = await self._client.request( method, @@ -493,11 +558,20 @@ async def _request( method, RATE_LIMIT_MAX_RETRIES, ) raise _rate_limited_error(response) - delay = _rate_limit_delay(response, attempt) + delay = _rate_limit_delay(response, attempt, waited) + if delay is None: + logger.warning( + "[DISPATCHARR] Rate limited (429) on %s; Retry-After exceeds the %.0fs wait budget, giving up", + method, RATE_LIMIT_MAX_TOTAL_WAIT, + ) + raise _rate_limited_error( + response, f"Retry-After exceeds the {RATE_LIMIT_MAX_TOTAL_WAIT:.0f}s wait budget", + ) logger.warning( "[DISPATCHARR] Rate limited (429) on %s; retrying in %.1fs (attempt %d/%d)", method, delay, attempt + 1, RATE_LIMIT_MAX_RETRIES, ) + waited += delay await _sleep(delay) if response.status_code >= 400: diff --git a/backend/routers/channel_pipeline.py b/backend/routers/channel_pipeline.py index f3e5b6a8e..af2651eea 100644 --- a/backend/routers/channel_pipeline.py +++ b/backend/routers/channel_pipeline.py @@ -2026,6 +2026,7 @@ async def commit_auto_creation_pipeline(request: CommitPipelinePlanRequest, _adm partial_replay = { "failed_index": exc.failed_index, "failed_write": exc.failed_write, + "failed_outcome": exc.failed_outcome, "completed_targets": exc.completed, "not_applied": exc.not_applied, "pre_mutation": exc.pre_mutation, @@ -2038,6 +2039,10 @@ async def commit_auto_creation_pipeline(request: CommitPipelinePlanRequest, _adm # would leave the caller unable to tell whether a retry is safe. # ``pre_mutation`` is True only when nothing landed AND the first # write was provably rejected (e.g. 429) before upstream mutated. + # ``failed_outcome`` distinguishes that rejection from a failure + # whose upstream effect is unknown (lost response, 5xx); + # ``not_applied`` lists only the writes that were never attempted. + # ``completed_writes`` is forward-call history, not current state. raise HTTPException( status_code=424, detail={ @@ -2045,6 +2050,7 @@ async def commit_auto_creation_pipeline(request: CommitPipelinePlanRequest, _adm "execution_id": execution_id, "failed_index": exc.failed_index, "failed_write": exc.failed_write, + "failed_outcome": exc.failed_outcome, "pre_mutation": exc.pre_mutation, "completed_writes": exc.completed, "not_applied": exc.not_applied, diff --git a/backend/services/pipeline_write_plan.py b/backend/services/pipeline_write_plan.py index 145bcb408..e94e0fdfc 100644 --- a/backend/services/pipeline_write_plan.py +++ b/backend/services/pipeline_write_plan.py @@ -86,20 +86,43 @@ def accounting(self) -> dict[str, int]: return {"write_count": len(self.writes), "unique_target_count": len(targets)} +# Outcome of the write at ``failed_index`` (GH #1009, PR #1010 review item 1). +FAILED_OUTCOME_REJECTED = "rejected" # upstream provably refused it before mutating +FAILED_OUTCOME_UNKNOWN = "unknown" # it may or may not have landed (lost response, 5xx, ...) + + class PartialReplayError(RuntimeError): """Upstream has no transaction; exposes exactly how far replay reached. - ``completed`` and ``not_applied`` are target strings (``method:first_arg``) - so a caller can see what landed and what did not. ``pre_mutation`` is - True only when NO write completed AND the failing write's exception proves - upstream rejected it before mutating anything (GH #1009); it is the one - case in which a fresh prepare + commit is known to be safe. + Three disjoint outcome classes are reported, so a caller can neither + repeat an operation that already landed nor skip one that never did: + + * ``completed`` — writes whose upstream call returned successfully, in + order. This is FORWARD-CALL HISTORY, not current upstream state: the + compensation pass may since have undone some of them, and an empty + ``compensation_errors`` says only that compensation raised nothing, + not that every effect was restored. + * ``failed_write`` — the write at ``failed_index``, with + ``failed_outcome`` saying whether upstream provably rejected it + (``"rejected"``: a 4xx response or a refused connection) or whether its + effect is ``"unknown"`` (timeout, dropped connection, 5xx, anything + else). An unknown outcome may have mutated upstream. + * ``not_applied`` — the writes AFTER the failed one. They were never + attempted. The failed write is deliberately NOT in this list. + + ``pre_mutation`` is True only when nothing completed, compensation was + clean, and the failed write was ``"rejected"``; it is the one case in + which a fresh prepare + commit is known to be safe. + + Target strings are bounded descriptors (see :func:`describe_write_target`): + ``method:`` where the id is known, otherwise a plan + operation index. They never include payload contents. """ def __init__( self, failed_index: int, completed: list[str], compensation_errors: list[str], *, failed_write: str = "", not_applied: list[str] | None = None, - pre_mutation: bool = False, + pre_mutation: bool = False, failed_outcome: str = FAILED_OUTCOME_UNKNOWN, ): super().__init__(f"pipeline replay failed at write {failed_index}") self.failed_index = failed_index @@ -108,10 +131,45 @@ def __init__( self.failed_write = failed_write self.not_applied = list(not_applied or []) self.pre_mutation = pre_mutation + self.failed_outcome = failed_outcome + +def describe_write_target( + write: PlannedWrite, index: int, remap: dict[int, int] | None = None, + *, created_id: int | None = None, +) -> str: + """Bounded, payload-free descriptor of a planned write's target. -def _write_target(write: PlannedWrite, args: list[Any]) -> str: - return f"{write.method}:{args[0] if args else ''}" + PR #1010 review items 2 and 3: the descriptor names the RESOLVED upstream + resource when the plan's temp id has already been mapped (``-1`` -> ``101``), + renders a still-unresolved future resource explicitly as ``pending()`` + instead of failing, and never stringifies argument dictionaries or URLs + (a ``create_logo`` payload can carry a credentialed URL). Writes whose + first argument is not an id (creates) are identified by plan position. + """ + remap = remap or {} + + def _id(value: Any) -> str: + if isinstance(value, bool) or not isinstance(value, int): + return f"#{index}" + if value < 0: + return str(remap[value]) if value in remap else f"pending({value})" + return str(value) + + if created_id is not None: + # A completed create: name the upstream resource it produced, so a + # failed compensation can still be located. + return f"{write.method}#{index}->{created_id}" + if not write.args: + return f"{write.method}#{index}" + first = write.args[0] + if isinstance(first, int) and not isinstance(first, bool): + return f"{write.method}:{_id(first)}" + if isinstance(first, list) and first and all( + isinstance(item, int) and not isinstance(item, bool) for item in first + ): + return f"{write.method}:[{','.join(_id(item) for item in first)}]" + return f"{write.method}#{index}" def _failure_precludes_mutation(exc: BaseException) -> bool: @@ -299,10 +357,15 @@ def mapped(value: Any) -> Any: next_temp = -1 completed: list[tuple[PlannedWrite, list[Any], Any]] = [] + # Set only once the failed write's call was actually issued; a mapping + # error before the call means upstream was never contacted. + attempted = False try: for write in plan.writes: + attempted = False args = mapped(write.args) kwargs = mapped(write.kwargs) + attempted = True if write.event_sync: from services.event_sync_cleanup import apply_change result = await apply_change(client, write.event_sync, execution_id) @@ -346,17 +409,36 @@ def mapped(value: Any) -> Any: }) except Exception as compensation_exc: # noqa: BLE001 compensation_errors.append(f"{done.method}: {compensation_exc}") - completed_targets = [_write_target(item[0], item[1]) for item in completed] failed_index = len(completed) - not_applied = [_write_target(write, write.args) for write in plan.writes[failed_index:]] + completed_targets = [ + describe_write_target( + item[0], position, remap, + created_id=( + item[2].get("id") if item[0].method.startswith("create_") + and isinstance(item[2], dict) else None + ), + ) + for position, item in enumerate(completed) + ] + failed_write = ( + describe_write_target(plan.writes[failed_index], failed_index, remap) + if failed_index < len(plan.writes) else "" + ) + # Writes AFTER the failed one were never attempted. The failed write + # itself is classified separately (rejected vs unknown), never as + # "not applied": a lost response may have landed upstream. + not_applied = [ + describe_write_target(write, position, remap) + for position, write in enumerate(plan.writes) + if position > failed_index + ] + rejected = (not attempted) or _failure_precludes_mutation(exc) raise PartialReplayError( failed_index, completed_targets, compensation_errors, - failed_write=not_applied[0] if not_applied else "", + failed_write=failed_write, not_applied=not_applied, - pre_mutation=( - not completed and not compensation_errors - and _failure_precludes_mutation(exc) - ), + failed_outcome=FAILED_OUTCOME_REJECTED if rejected else FAILED_OUTCOME_UNKNOWN, + pre_mutation=(not completed and not compensation_errors and rejected), ) from exc return results, remap diff --git a/backend/tests/routers/test_channel_pipeline.py b/backend/tests/routers/test_channel_pipeline.py index a303b5071..2079fc291 100644 --- a/backend/tests/routers/test_channel_pipeline.py +++ b/backend/tests/routers/test_channel_pipeline.py @@ -5430,7 +5430,8 @@ async def test_partial_replay_returns_424_with_applied_and_not_applied(self, tes ) failure = PartialReplayError( 0, [], [], failed_write="update_channel:7", - not_applied=["update_channel:7", "delete_channel:8"], pre_mutation=True, + not_applied=["delete_channel:8"], pre_mutation=True, + failed_outcome="rejected", ) marked: dict = {} @@ -5454,9 +5455,14 @@ def capture_failed(execution_id, error, *, partial_replay=None): assert detail["failed_index"] == 0 assert detail["failed_write"] == "update_channel:7" assert detail["pre_mutation"] is True + assert detail["failed_outcome"] == "rejected" assert detail["completed_writes"] == [] - assert detail["not_applied"] == ["update_channel:7", "delete_channel:8"] + # The failed write is classified by failed_outcome, never listed as + # "not applied": only the writes after it were never attempted. + assert detail["not_applied"] == ["delete_channel:8"] assert detail["compensation_errors"] == [] assert marked["execution_id"] == detail["execution_id"] assert marked["partial_replay"]["failed_write"] == "update_channel:7" + assert marked["partial_replay"]["failed_outcome"] == "rejected" + assert marked["partial_replay"]["not_applied"] == ["delete_channel:8"] assert marked["partial_replay"]["pre_mutation"] is True diff --git a/backend/tests/services/test_pipeline_write_plan.py b/backend/tests/services/test_pipeline_write_plan.py index 74128da11..a35821b54 100644 --- a/backend/tests/services/test_pipeline_write_plan.py +++ b/backend/tests/services/test_pipeline_write_plan.py @@ -136,8 +136,9 @@ async def test_first_write_rejected_with_429_is_reported_as_pre_mutation(): exc = error.value assert exc.failed_index == 0 assert exc.failed_write == "update_channel:7" + assert exc.failed_outcome == "rejected" assert exc.completed == [] - assert exc.not_applied == ["update_channel:7", "delete_channel:8"] + assert exc.not_applied == ["delete_channel:8"] assert exc.pre_mutation is True live.delete_channel.assert_not_awaited() @@ -152,8 +153,9 @@ async def test_failure_after_a_landed_write_is_not_pre_mutation(): exc = error.value assert exc.failed_index == 1 assert exc.failed_write == "delete_channel:8" + assert exc.failed_outcome == "rejected" assert exc.completed == ["update_channel:7"] - assert exc.not_applied == ["delete_channel:8"] + assert exc.not_applied == [] assert exc.pre_mutation is False @@ -167,3 +169,155 @@ async def test_first_write_timeout_is_not_claimed_pre_mutation(): assert error.value.failed_index == 0 assert error.value.completed == [] assert error.value.pre_mutation is False + assert error.value.failed_outcome == "unknown" + + +# --------------------------------------------------------------------------- +# PR #1010 review items 1-3: outcome classes, resolved ids, safe descriptors. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_landed_write_with_lost_response_is_unknown_not_not_applied(): + """Item 1: a PATCH that reached upstream but whose response was lost may + have landed. It must be reported as failed with an unknown outcome and + must NOT appear in ``not_applied``; only the never-attempted delete is.""" + live = AsyncMock() + live.update_channel.side_effect = httpx.RemoteProtocolError( + "peer closed connection", request=httpx.Request("PATCH", "http://d/x"), + ) + with pytest.raises(PartialReplayError) as error: + await replay_write_plan(live, _two_write_plan()) + exc = error.value + assert exc.failed_write == "update_channel:7" + assert exc.failed_outcome == "unknown" + assert exc.not_applied == ["delete_channel:8"] + assert "update_channel:7" not in exc.not_applied + assert exc.pre_mutation is False + live.delete_channel.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_confirmed_rejection_control_is_rejected_and_pre_mutation(): + live = AsyncMock() + live.update_channel.side_effect = httpx.HTTPStatusError( + "400", request=httpx.Request("PATCH", "http://d/x"), + response=httpx.Response(400, request=httpx.Request("PATCH", "http://d/x")), + ) + with pytest.raises(PartialReplayError) as error: + await replay_write_plan(live, _two_write_plan()) + assert error.value.failed_outcome == "rejected" + assert error.value.pre_mutation is True + assert error.value.not_applied == ["delete_channel:8"] + + +@pytest.mark.asyncio +async def test_dependent_update_failure_reports_the_resolved_upstream_id(): + """Item 2: the plan recorded the update against temp id -1; replay created + channel 101 and PATCHed 101, so the failure must name 101, and the + completed create must name the resource it produced.""" + live = AsyncMock() + live.create_channel.return_value = {"id": 101} + live.update_channel.side_effect = _rate_limited() + plan = PipelineWritePlan(writes=[ + PlannedWrite("create_channel", [{"name": "New"}], {}), + PlannedWrite("update_channel", [-1, {"streams": [5]}], {}), + PlannedWrite("update_channel", [-1, {"name": "Renamed"}], {}), + ]) + with pytest.raises(PartialReplayError) as error: + await replay_write_plan(live, plan) + exc = error.value + assert exc.failed_index == 1 + assert exc.completed == ["create_channel#0->101"] + assert exc.failed_write == "update_channel:101" + assert exc.not_applied == ["update_channel:101"] + assert "-1" not in exc.failed_write and "-1" not in "".join(exc.not_applied) + live.delete_channel.assert_awaited_once_with(101) # compensation targeted the real id + + +@pytest.mark.asyncio +async def test_compensation_failure_names_the_created_resource(): + live = AsyncMock() + live.create_channel.return_value = {"id": 101} + live.update_channel.side_effect = _rate_limited() + live.delete_channel.side_effect = RuntimeError("upstream unavailable") + plan = PipelineWritePlan(writes=[ + PlannedWrite("create_channel", [{"name": "New"}], {}), + PlannedWrite("update_channel", [-1, {"streams": [5]}], {}), + ]) + with pytest.raises(PartialReplayError) as error: + await replay_write_plan(live, plan) + assert error.value.completed == ["create_channel#0->101"] + assert error.value.compensation_errors == ["create_channel: upstream unavailable"] + assert error.value.pre_mutation is False + + +@pytest.mark.asyncio +async def test_unresolved_future_target_renders_as_pending_without_failing(): + live = AsyncMock() + live.update_channel.side_effect = _rate_limited() + plan = PipelineWritePlan(writes=[ + PlannedWrite("update_channel", [7, {"name": "A"}], {}), + PlannedWrite("create_channel_group", ["Sports"], {}), + PlannedWrite("create_channel", [{"name": "New"}], {}), + PlannedWrite("update_channel", [-2, {"streams": [5]}], {}), + PlannedWrite("assign_channel_numbers", [[7, -2], 100], {}), + ]) + with pytest.raises(PartialReplayError) as error: + await replay_write_plan(live, plan) + assert error.value.not_applied == [ + "create_channel_group#1", "create_channel#2", + "update_channel:pending(-2)", "assign_channel_numbers:[7,pending(-2)]", + ] + + +@pytest.mark.asyncio +async def test_credential_bearing_create_payload_never_reaches_diagnostics(): + """Item 3: a create_logo payload URL can carry a provider token. No + diagnostic field, and not str(exc), may contain it.""" + token = "SECRET-TOKEN-8f3a9c" + live = AsyncMock() + live.create_logo.side_effect = _rate_limited() + plan = PipelineWritePlan(writes=[ + PlannedWrite("create_logo", [{"name": "L", "url": f"http://p.example/logo.png?token={token}"}], {}), + PlannedWrite("create_channel", [{"name": "New", "tvg_id": token}], {}), + PlannedWrite("update_channel", [-2, {"logo_id": -1}], {}), + ]) + with pytest.raises(PartialReplayError) as error: + await replay_write_plan(live, plan) + exc = error.value + assert exc.failed_write == "create_logo#0" + assert exc.not_applied == ["create_channel#1", "update_channel:pending(-2)"] + assert exc.pre_mutation is True + rendered = " ".join([exc.failed_write, *exc.not_applied, *exc.completed, str(exc), repr(exc)]) + assert token not in rendered + assert "http" not in rendered + + +@pytest.mark.asyncio +async def test_completed_create_descriptor_is_payload_free(): + token = "SECRET-TOKEN-8f3a9c" + live = AsyncMock() + live.create_logo.return_value = {"id": 55} + live.create_channel.side_effect = _rate_limited() + plan = PipelineWritePlan(writes=[ + PlannedWrite("create_logo", [{"name": "L", "url": f"http://p.example/logo.png?token={token}"}], {}), + PlannedWrite("create_channel", [{"name": "New"}], {}), + ]) + with pytest.raises(PartialReplayError) as error: + await replay_write_plan(live, plan) + assert error.value.completed == ["create_logo#0->55"] + assert token not in " ".join(error.value.completed) + + +@pytest.mark.asyncio +async def test_mapping_error_before_the_call_is_a_rejection_not_unknown(): + """An unresolved temp id fails inside replay before upstream is contacted.""" + live = AsyncMock() + plan = PipelineWritePlan(writes=[PlannedWrite("update_channel", [-9, {"name": "A"}], {})]) + with pytest.raises(PartialReplayError) as error: + await replay_write_plan(live, plan) + assert error.value.failed_outcome == "rejected" + assert error.value.failed_write == "update_channel:pending(-9)" + assert error.value.pre_mutation is True + live.update_channel.assert_not_awaited() diff --git a/backend/tests/unit/test_dispatcharr_client_rate_limit.py b/backend/tests/unit/test_dispatcharr_client_rate_limit.py index d8497e76f..db4ca8717 100644 --- a/backend/tests/unit/test_dispatcharr_client_rate_limit.py +++ b/backend/tests/unit/test_dispatcharr_client_rate_limit.py @@ -124,3 +124,182 @@ async def test_login_budget_failure_leaves_no_token(): finally: await client._client.aclose() assert client.access_token is None + + +# --------------------------------------------------------------------------- +# PR #1010 review items 4 and 5: finite, validated Retry-After handling under +# an explicit total wait budget, in both the delay-seconds and HTTP-date forms. +# --------------------------------------------------------------------------- + +from email.utils import format_datetime +from datetime import datetime, timedelta, timezone + +from dispatcharr_client import RATE_LIMIT_MAX_TOTAL_WAIT, _parse_retry_after, _rate_limit_delay + +_FIXED_NOW = 1_800_000_000.0 + + +def _http_date(offset_seconds: float) -> str: + when = datetime.fromtimestamp(_FIXED_NOW + offset_seconds, tz=timezone.utc) + return format_datetime(when, usegmt=True) + + +class TestParseRetryAfter: + def test_delay_seconds(self): + assert _parse_retry_after("7") == 7.0 + assert _parse_retry_after(" 0 ") == 0.0 + assert _parse_retry_after("2.5") == 2.5 + + @pytest.mark.parametrize("value", ["inf", "1e309", "nan", "-5", "abc", "", None]) + def test_non_finite_negative_and_garbage_are_unusable(self, value): + assert _parse_retry_after(value) is None + + def test_http_date_in_the_future_is_the_remaining_interval(self): + with patch.object(dispatcharr_client, "_now", lambda: _FIXED_NOW): + remaining = _parse_retry_after(_http_date(20)) + assert 19.0 <= remaining <= 20.0 # HTTP dates have 1s resolution + + def test_http_date_in_the_past_means_retry_now(self): + with patch.object(dispatcharr_client, "_now", lambda: _FIXED_NOW): + assert _parse_retry_after(_http_date(-90)) == 0.0 + + def test_delay_over_budget_is_reported_as_no_admissible_wait(self): + response = _response(429, headers={"Retry-After": "86400"}) + assert _rate_limit_delay(response, 0, 0.0) is None + + def test_cumulative_waits_are_bounded_by_the_budget(self): + response = _response(429, headers={"Retry-After": "20"}) + assert _rate_limit_delay(response, 0, 0.0) == 20.0 + assert _rate_limit_delay(response, 1, 20.0) is None # 40 > 30 + + def test_backoff_fallback_is_also_subject_to_the_budget(self): + response = _response(429) + assert _rate_limit_delay(response, 0, RATE_LIMIT_MAX_TOTAL_WAIT - 0.5) is None + + +@pytest.mark.asyncio +async def test_request_gives_up_immediately_on_a_retry_after_beyond_the_budget(): + client = _api_key_client() + sleeper = AsyncMock() + try: + client._client.request = AsyncMock(return_value=_response(429, headers={"Retry-After": "86400"})) + with patch.object(dispatcharr_client, "_sleep", sleeper): + with pytest.raises(httpx.HTTPStatusError) as error: + await client._request("GET", "/api/x/") + finally: + await client._client.aclose() + sleeper.assert_not_awaited() + assert client._client.request.await_count == 1 + assert error.value.response.status_code == 429 + assert "wait budget" in str(error.value) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("header", ["inf", "nan", "-5", "garbage"]) +async def test_request_falls_back_to_backoff_for_invalid_retry_after(header): + client = _api_key_client() + sleeps: list[float] = [] + try: + client._client.request = AsyncMock(side_effect=[ + _response(429, headers={"Retry-After": header}), _response(200), + ]) + with patch.object(dispatcharr_client, "_sleep", AsyncMock(side_effect=sleeps.append)): + await client._request("GET", "/api/x/") + finally: + await client._client.aclose() + assert sleeps == [1.0] + + +@pytest.mark.asyncio +async def test_request_honours_http_date_retry_after(): + client = _api_key_client() + sleeps: list[float] = [] + try: + client._client.request = AsyncMock(side_effect=[ + _response(429, headers={"Retry-After": _http_date(20)}), _response(200), + ]) + with patch.object(dispatcharr_client, "_sleep", AsyncMock(side_effect=sleeps.append)), \ + patch.object(dispatcharr_client, "_now", lambda: _FIXED_NOW): + await client._request("GET", "/api/x/") + finally: + await client._client.aclose() + assert len(sleeps) == 1 and 19.0 <= sleeps[0] <= 20.0 + + +@pytest.mark.asyncio +async def test_request_http_date_beyond_budget_is_terminal_not_an_early_retry(): + client = _api_key_client() + sleeper = AsyncMock() + try: + client._client.request = AsyncMock(return_value=_response(429, headers={"Retry-After": _http_date(600)})) + with patch.object(dispatcharr_client, "_sleep", sleeper), \ + patch.object(dispatcharr_client, "_now", lambda: _FIXED_NOW): + with pytest.raises(httpx.HTTPStatusError) as error: + await client._request("GET", "/api/x/") + finally: + await client._client.aclose() + sleeper.assert_not_awaited() + assert client._client.request.await_count == 1 + assert error.value.response.status_code == 429 + + +@pytest.mark.asyncio +async def test_request_cumulative_retry_after_waits_stop_at_the_budget(): + client = _api_key_client() + sleeps: list[float] = [] + try: + client._client.request = AsyncMock(return_value=_response(429, headers={"Retry-After": "20"})) + with patch.object(dispatcharr_client, "_sleep", AsyncMock(side_effect=sleeps.append)): + with pytest.raises(httpx.HTTPStatusError): + await client._request("GET", "/api/x/") + finally: + await client._client.aclose() + assert sleeps == [20.0] # a second 20s wait would exceed the 30s budget + assert client._client.request.await_count == 2 + + +@pytest.mark.asyncio +async def test_login_over_budget_retry_after_releases_the_auth_lock(): + """The login path waits while holding ``_auth_lock``; an unbounded wait + would stall every request behind it. Over budget must raise promptly and + leave the lock free.""" + client = _jwt_client() + sleeper = AsyncMock() + try: + client._client.post = AsyncMock(return_value=_response(429, headers={"Retry-After": "inf"})) + # "inf" is invalid -> backoff; make the budget already exhausted so + # the first backoff is refused, then check the same for a huge number. + with patch.object(dispatcharr_client, "_sleep", sleeper), \ + patch.object(dispatcharr_client, "RATE_LIMIT_MAX_TOTAL_WAIT", 0.5): + with pytest.raises(httpx.HTTPStatusError) as error: + await client._ensure_authenticated() + assert not client._auth_lock.locked() + assert client.access_token is None + assert error.value.response.status_code == 429 + + client._client.post = AsyncMock(return_value=_response(429, headers={"Retry-After": "86400"})) + with patch.object(dispatcharr_client, "_sleep", sleeper): + with pytest.raises(httpx.HTTPStatusError): + await client._ensure_authenticated() + assert not client._auth_lock.locked() + finally: + await client._client.aclose() + sleeper.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_login_honours_http_date_retry_after_within_budget(): + client = _jwt_client() + sleeps: list[float] = [] + try: + client._client.post = AsyncMock(side_effect=[ + _response(429, headers={"Retry-After": _http_date(5)}), + _response(200, json_body={"access": "A", "refresh": "R"}), + ]) + with patch.object(dispatcharr_client, "_sleep", AsyncMock(side_effect=sleeps.append)), \ + patch.object(dispatcharr_client, "_now", lambda: _FIXED_NOW): + await client._login() + finally: + await client._client.aclose() + assert len(sleeps) == 1 and 4.0 <= sleeps[0] <= 5.0 + assert client.access_token == "A" diff --git a/docs/user_guide/integrations/mcp.md b/docs/user_guide/integrations/mcp.md index 222dd7769..6682dfeeb 100644 --- a/docs/user_guide/integrations/mcp.md +++ b/docs/user_guide/integrations/mcp.md @@ -741,14 +741,32 @@ decision or exact write payload requires a new preview. ECM persists the exact execution program and target-scoped rollback snapshot immediately before replay, then compensates reversible writes in reverse order if a later write fails. A partial failure returns `424 Failed Dependency` (not a proxy-style 502) whose -body names the write that failed, lists target-specific completed writes and -the writes that were not applied, reports any compensation failures, and -carries the execution id to inspect. Its `pre_mutation` flag is `true` only -when nothing was applied and Dispatcharr provably rejected the first write -before mutating anything (for example a rate-limit rejection); that is the one -case in which a fresh prepare and commit is known to be safe. Deletes and -channel-profile membership changes cannot always be restored with the same -upstream identifiers; inspect the execution record and rollback snapshot before +body carries the execution id to inspect and sorts every planned write into +one of three classes: + +* `completed_writes` — writes whose Dispatcharr call returned successfully, + in order. This is forward-call history, not current upstream state: the + compensation pass may since have undone some of them, and an empty + `compensation_errors` means only that compensation raised nothing. +* `failed_write` with `failed_outcome` — the write at `failed_index`. + `"rejected"` means Dispatcharr provably refused it before mutating anything + (a 4xx answer such as a rate limit, or a refused connection). `"unknown"` + means its effect cannot be established (a timeout, a dropped connection, a + 5xx): the write **may have landed**, so inspect Dispatcharr before repeating + it. +* `not_applied` — the writes after the failed one. They were never attempted. + +`pre_mutation` is `true` only when nothing completed, compensation was clean +and the failed write was rejected; that is the one case in which a fresh +prepare and commit is known to be safe. Write descriptors name the resolved +Dispatcharr id where one is known (`update_channel:101`, even when the plan +recorded it as a temporary id), show a not-yet-created target as +`pending(-1)`, and identify creates by plan position (`create_logo#0`); they +never include payload contents such as logo URLs. Deletes and channel-profile +membership changes cannot always be restored with the same upstream +identifiers; inspect the execution record and rollback snapshot before retrying a partially failed commit. Transient `429 Too Many Requests` answers -from Dispatcharr are retried with bounded backoff before they count as a -failure. +from Dispatcharr are retried with bounded exponential backoff (at most three +retries, honouring a `Retry-After` given as seconds or as an HTTP date, and +never more than 30 seconds of total waiting per call); a `Retry-After` beyond +that budget is surfaced as a rate-limit failure rather than waited on. From 1ebdb37c933e884f0130c13ffa7aaa9e83891a57 Mon Sep 17 00:00:00 2001 From: lukeeexd <31347888+lukeeexd@users.noreply.github.com> Date: Sun, 20 Sep 2026 18:28:50 +0100 Subject: [PATCH 3/4] test(channel-pipeline): cross-boundary partial-replay evidence for the #1010 review (#1009) Strict-delta review on PR #1010 accepted items 4 and 5 and asked for retained regressions proving items 1-3 across the HTTP, persistence and MCP boundaries rather than at PartialReplayError. - New backend/tests/integration/test_gh1009_commit_partial_failure_evidence.py drives the real DispatcharrClient over a controlled httpx.MockTransport that plays Dispatcharr, the real replay_write_plan and compensation pass, the real POST /api/auto-creation/run/commit handler over ASGI, and the real _mark_execution_failed persistence. Item 1: a PATCH that lands upstream and then loses its response yields failed_outcome=unknown, the failed write absent from not_applied and the later delete never attempted, in the 424 body, the execution log entry and the snapshot recovery evidence, with a 429 confirmed-rejection control (rejected, pre_mutation=true) through the same boundary. Item 2: create -> dependent update rejected reports completed create_channel#0->101 and failed_write update_channel:101 with no temp id anywhere in the response or persisted evidence, and compensation DELETEs the real id; a compensation-failure variant (DELETE 503) keeps the orphan locatable in both the 424 and the persisted evidence. Item 3: a create_logo payload carrying a canary token in its URL, and a create_channel carrying it in tvg_id, leave no trace in the 424 body, the persisted partial-failure fields, the execution error message or the backend DEBUG log, while execution_id and the plan-position descriptors survive. - New mcp-server/tests/test_gh1009_partial_failure_presentation.py runs the sidecar's real ECMClient.post and the run_channel_pipeline tool over a mock transport answering the backend's 424 shape, asserting the operator text and the sidecar log carry execution_id, failed_write, failed_outcome and not_applied, plus a control showing the sidecar is a verbatim pass-through, which is why the payload-free guarantee is asserted at the producer. - Rebased onto origin/dev (#1006, #1008, #1017 landed; dev is 0036) and bumped the three version touchpoints and the changelog to 0.18.2-0039. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 2 +- backend/main.py | 2 +- backend/routers/backup.py | 2 +- ..._gh1009_commit_partial_failure_evidence.py | 350 ++++++++++++++++++ frontend/package.json | 2 +- ...est_gh1009_partial_failure_presentation.py | 135 +++++++ 6 files changed, 489 insertions(+), 4 deletions(-) create mode 100644 backend/tests/integration/test_gh1009_commit_partial_failure_evidence.py create mode 100644 mcp-server/tests/test_gh1009_partial_failure_presentation.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2226b4bc4..c128564a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Fixed - **Event Sync pre-flight names the real constraint when the master group is not M3U-backed ([#1007](https://github.com/MotWakorb/enhancedchannelmanager/issues/1007); build 0.18.2-0036).** The `group_settings_found` failure for a master group used to read as a lookup error ("was not found in the M3U account's group settings — the group may have been removed or renamed"), which sent operators looking for a missing group when the group existed but was hand-curated. The message now states that the master must come from an M3U account with `auto_channel_sync` ON because Dispatcharr owns master-channel lifecycle, that a hand-curated channel group is not supported as a master, and only then mentions the removed/renamed case. The check id, failure shape and behaviour are unchanged; `docs/event_sync.md` gains the same note under "Pick the master group" and "Pre-flight checks". The diagnosis is scoped to what was actually checked: a provider-scoped master reports the missing provider/group association (and whether another account carries the group) instead of claiming no account carries it, and "M3U-backed" includes a whole-group Channel Group Override target of an auto-synced group. -- **Planned pipeline commit reports a partial failure honestly and survives Dispatcharr rate limiting (GitHub #1009; build 0.18.2-0036).** `POST /api/channel-pipeline/run/commit` returned `502` with an empty `completed_writes` list whenever the first replayed write failed, which a caller could not tell apart from a gateway failure. A partial replay now returns `424 Failed Dependency` with the execution id, the writes that completed (forward-call history), the failed write with a `failed_outcome` of `rejected` (upstream provably refused it) or `unknown` (a lost response or 5xx that may have landed), the later writes that were never attempted, compensation errors, and a `pre_mutation` flag that is true only when nothing landed and the failed write was rejected. Write descriptors name the resolved Dispatcharr id where known, mark a not-yet-created target as pending, and never include payload contents such as logo URLs. The Dispatcharr client retries `429 Too Many Requests` on API requests and login with bounded exponential backoff, honouring `Retry-After` in both its seconds and HTTP-date forms, within an explicit 30-second total wait budget per call; a `Retry-After` beyond the budget, or a spent retry budget, surfaces as an HTTP status error carrying the 429 instead of an unbounded wait. Previously no code path handled a 429 and a single one aborted the replay. The commit endpoint remains synchronous; the 30s request-timeout exposure raised in #1009 is left for a separate decision. +- **Planned pipeline commit reports a partial failure honestly and survives Dispatcharr rate limiting (GitHub #1009; build 0.18.2-0039).** `POST /api/channel-pipeline/run/commit` returned `502` with an empty `completed_writes` list whenever the first replayed write failed, which a caller could not tell apart from a gateway failure. A partial replay now returns `424 Failed Dependency` with the execution id, the writes that completed (forward-call history), the failed write with a `failed_outcome` of `rejected` (upstream provably refused it) or `unknown` (a lost response or 5xx that may have landed), the later writes that were never attempted, compensation errors, and a `pre_mutation` flag that is true only when nothing landed and the failed write was rejected. Write descriptors name the resolved Dispatcharr id where known, mark a not-yet-created target as pending, and never include payload contents such as logo URLs. The Dispatcharr client retries `429 Too Many Requests` on API requests and login with bounded exponential backoff, honouring `Retry-After` in both its seconds and HTTP-date forms, within an explicit 30-second total wait budget per call; a `Retry-After` beyond the budget, or a spent retry budget, surfaces as an HTTP status error carrying the 429 instead of an unbounded wait. Previously no code path handled a 429 and a single one aborted the replay. The commit endpoint remains synchronous; the 30s request-timeout exposure raised in #1009 is left for a separate decision. - **Section navigation highlights the settled reading position (bead `enhancedchannelmanager-2896r.19`; build 0.18.2-0035).** Scheduled Tasks keeps the chosen section highlighted after scrolling in either direction, including bottom-clamped targets and reloaded links. Refresh preserves the reader's position without replaying a consumed section link. diff --git a/backend/main.py b/backend/main.py index f549c004f..633f91683 100644 --- a/backend/main.py +++ b/backend/main.py @@ -161,7 +161,7 @@ Login endpoints are rate-limited to 5 requests per minute per IP address. """, - version="0.18.2-0036", + version="0.18.2-0039", openapi_tags=tags_metadata, docs_url="/api/docs", redoc_url="/api/redoc", diff --git a/backend/routers/backup.py b/backend/routers/backup.py index b4d14b2c3..00de6849d 100644 --- a/backend/routers/backup.py +++ b/backend/routers/backup.py @@ -133,7 +133,7 @@ def _resolve_backup_normalization_group_ids(item: dict, session) -> str | None: # scripts/check_version_consistency.py that used to fail the PR on divergence # were removed. Do NOT rename it, change its shape, or repurpose it. It is an INFORMATIONAL human-readable string ("which # ECM build produced this artifact") — it is NOT a compatibility gate. -APP_VERSION = "0.18.2-0036" +APP_VERSION = "0.18.2-0039" # DBAS backup-artifact schema version (ADR-008 D1 / ADR-012 D1). This is a # DEDICATED, MONOTONIC INTEGER that is DISTINCT from the human-readable diff --git a/backend/tests/integration/test_gh1009_commit_partial_failure_evidence.py b/backend/tests/integration/test_gh1009_commit_partial_failure_evidence.py new file mode 100644 index 000000000..29a19f186 --- /dev/null +++ b/backend/tests/integration/test_gh1009_commit_partial_failure_evidence.py @@ -0,0 +1,350 @@ +"""GH #1009 / PR #1010: partial-replay evidence across the real boundaries. + +The replay unit tests prove the classification on ``PartialReplayError``; +the router test injects a prebuilt error. Neither can see a regression in +how the outcome travels from the upstream socket to the 424 body and into +SQLite. This module drives that whole path with real code: + +* the real ``DispatcharrClient`` over a controlled ``httpx.MockTransport`` + that plays Dispatcharr (and can apply a PATCH and then drop the + response, or answer 4xx/5xx); +* the real ``replay_write_plan`` and its compensation pass; +* the real ``POST /api/auto-creation/run/commit`` handler over ASGI, so + the assertions are on the HTTP response the MCP consumer receives; +* the real ``_mark_execution_failed`` persistence into the + ``ChannelPipelineExecution`` row and the ``ChannelPipelineSnapshot`` + recovery evidence. + +Only the plan computation is doubled (it returns the stored plan +unchanged, which is what a non-drifted commit sees). +""" +from __future__ import annotations + +import json +import logging +from unittest.mock import AsyncMock, patch + +import httpx +import pytest +from sqlalchemy.orm import sessionmaker + +from config import DispatcharrSettings +from dispatcharr_client import DispatcharrClient +from models import ChannelPipelineExecution, ChannelPipelineSnapshot +from routers import channel_pipeline as router +from services import mutation_plan_store as store +from services.mutation_plan_store import canonical_hash + +COMMIT_URL = "/api/auto-creation/run/commit" +CANARY = "SECRET-TOKEN-8f3a9c-canary" + + +class Upstream: + """A scripted Dispatcharr behind ``httpx.MockTransport``. + + ``channels`` is the upstream state; ``script`` maps ``(METHOD, path)`` + to a behaviour: a callable receiving the request and returning an + ``httpx.Response`` or raising an ``httpx`` transport error. Every request + is appended to ``calls`` so a test can prove what upstream received. + """ + + def __init__(self) -> None: + self.channels: dict[int, dict] = {} + self.calls: list[tuple[str, str]] = [] + self.script: dict[tuple[str, str], object] = {} + self.next_id = 101 + + def handler(self, request: httpx.Request) -> httpx.Response: + key = (request.method, request.url.path) + self.calls.append(key) + behaviour = self.script.get(key) + if behaviour is not None: + return behaviour(request) + return self.default(request) + + def default(self, request: httpx.Request) -> httpx.Response: + path = request.url.path + if request.method == "POST" and path == "/api/channels/channels/": + body = json.loads(request.content or b"{}") + new_id, self.next_id = self.next_id, self.next_id + 1 + self.channels[new_id] = {"id": new_id, **body} + return httpx.Response(201, json=self.channels[new_id]) + if request.method == "POST" and path == "/api/channels/logos/": + return httpx.Response(201, json={"id": 55}) + if path.startswith("/api/channels/channels/") and path.endswith("/"): + channel_id = int(path.rstrip("/").rsplit("/", 1)[-1]) + if request.method == "PATCH": + body = json.loads(request.content or b"{}") + self.channels.setdefault(channel_id, {"id": channel_id}).update(body) + return httpx.Response(200, json=self.channels[channel_id]) + if request.method == "DELETE": + self.channels.pop(channel_id, None) + return httpx.Response(204) + if request.method == "GET": + return httpx.Response(200, json=self.channels.get(channel_id, {"id": channel_id})) + raise AssertionError(f"unexpected upstream call: {request.method} {path}") + + def apply_then_lose_response(self, request: httpx.Request) -> httpx.Response: + """The PATCH lands upstream; the connection dies before the reply.""" + self.default(request) + raise httpx.RemoteProtocolError("peer closed connection without sending a response") + + @staticmethod + def reject(status: int): + def _reject(request: httpx.Request) -> httpx.Response: + return httpx.Response(status, json={"detail": "rejected by upstream"}) + return _reject + + +def _client(upstream: Upstream) -> DispatcharrClient: + client = DispatcharrClient( + DispatcharrSettings(url="http://dispatcharr", auth_method="api_key", api_key="k") + ) + client._client = httpx.AsyncClient(transport=httpx.MockTransport(upstream.handler)) + return client + + +def _payload(writes: list[dict], snapshot: list[dict] | None = None) -> dict: + return { + "request": {"m3u_account_ids": None, "rule_ids": [7]}, + "result": {"event_sync": [], "planned_review_candidates": [], "execution_log": []}, + "write_plan": { + "writes": [{"kwargs": {}, "event_sync": None, **w} for w in writes], + "channel_preconditions": {}, "group_preconditions": {}, "profile_preconditions": {}, + }, + "snapshot": snapshot or [], + } + + +async def _commit(async_client, test_engine, upstream: Upstream, payload: dict) -> tuple[httpx.Response, DispatcharrClient]: + fresh_store = store.MutationPlanStore() + plan = fresh_store.create( + "channel_pipeline", payload, canonical_hash(router._canonical_pipeline_decision(payload)), + ) + client = _client(upstream) + engine = type("Engine", (), {"client": client})() + try: + with patch.object(store, "mutation_plan_store", fresh_store), \ + patch.object(router, "_ensure_engine", AsyncMock(return_value=engine)), \ + patch.object(router, "_compute_pipeline_plan_payload", AsyncMock(return_value=payload)), \ + patch.object(router, "get_session", sessionmaker(bind=test_engine)): + response = await async_client.post(COMMIT_URL, json={ + "plan_id": plan.plan_id, "plan_hash": plan.payload_hash, "phase": "execute", + }) + finally: + await client._client.aclose() + return response, client + + +def _persisted(test_engine, execution_id: int) -> tuple[ChannelPipelineExecution, dict, dict]: + session = sessionmaker(bind=test_engine)() + try: + execution = session.get(ChannelPipelineExecution, execution_id) + log = execution.get_execution_log() + snapshot = session.query(ChannelPipelineSnapshot).filter_by(execution_id=execution_id).one() + return execution, log[0] if log else {}, snapshot.get_channels_data().get("partial_replay", {}) + finally: + session.close() + + +def _everywhere(response: httpx.Response, log_entry: dict, evidence: dict, execution: ChannelPipelineExecution) -> str: + return " ".join([ + response.text, json.dumps(log_entry), json.dumps(evidence), + execution.error_message or "", + ]) + + +# --------------------------------------------------------------------------- +# Item 1: unknown vs rejected vs unattempted, end to end. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_landed_write_with_lost_response_is_unknown_in_http_and_persistence( + async_client, test_engine, +): + upstream = Upstream() + upstream.channels = {7: {"id": 7, "name": "Old"}, 8: {"id": 8, "name": "Doomed"}} + upstream.script[("PATCH", "/api/channels/channels/7/")] = upstream.apply_then_lose_response + payload = _payload([ + {"method": "update_channel", "args": [7, {"name": "New"}]}, + {"method": "delete_channel", "args": [8]}, + ]) + + response, _ = await _commit(async_client, test_engine, upstream, payload) + + assert response.status_code == 424, response.text + detail = response.json()["detail"] + # Upstream DID change channel 7 — the inventory must not say otherwise. + assert upstream.channels[7]["name"] == "New" + assert 8 in upstream.channels # the delete was never attempted + assert ("DELETE", "/api/channels/channels/8/") not in upstream.calls + assert detail["failed_write"] == "update_channel:7" + assert detail["failed_outcome"] == "unknown" + assert detail["not_applied"] == ["delete_channel:8"] + assert "update_channel:7" not in detail["not_applied"] + assert detail["completed_writes"] == [] + assert detail["pre_mutation"] is False + assert detail["compensation_errors"] == [] + + execution, log_entry, evidence = _persisted(test_engine, detail["execution_id"]) + assert execution.status == "failed" + assert log_entry["type"] == "partial_replay_failure" + for persisted in (log_entry, evidence): + assert persisted["failed_write"] == "update_channel:7" + assert persisted["failed_outcome"] == "unknown" + assert persisted["not_applied"] == ["delete_channel:8"] + assert persisted["pre_mutation"] is False + assert persisted["completed_targets"] == [] + + +@pytest.mark.asyncio +async def test_confirmed_rejection_control_is_rejected_and_pre_mutation_everywhere( + async_client, test_engine, +): + upstream = Upstream() + upstream.channels = {7: {"id": 7, "name": "Old"}, 8: {"id": 8}} + upstream.script[("PATCH", "/api/channels/channels/7/")] = Upstream.reject(429) + payload = _payload([ + {"method": "update_channel", "args": [7, {"name": "New"}]}, + {"method": "delete_channel", "args": [8]}, + ]) + + with patch("dispatcharr_client._sleep", new=AsyncMock()): + response, _ = await _commit(async_client, test_engine, upstream, payload) + + assert response.status_code == 424, response.text + detail = response.json()["detail"] + assert upstream.channels[7]["name"] == "Old" # provably not applied + assert detail["failed_write"] == "update_channel:7" + assert detail["failed_outcome"] == "rejected" + assert detail["pre_mutation"] is True + assert detail["not_applied"] == ["delete_channel:8"] + + execution, log_entry, evidence = _persisted(test_engine, detail["execution_id"]) + assert execution.status == "failed" + for persisted in (log_entry, evidence): + assert persisted["failed_outcome"] == "rejected" + assert persisted["pre_mutation"] is True + assert persisted["not_applied"] == ["delete_channel:8"] + + +# --------------------------------------------------------------------------- +# Item 2: the resolved upstream id, through commit persistence, including a +# compensation failure where locating the remaining resource matters. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_dependent_write_failure_names_the_real_created_id_in_http_and_persistence( + async_client, test_engine, +): + upstream = Upstream() + upstream.script[("PATCH", "/api/channels/channels/101/")] = Upstream.reject(400) + payload = _payload([ + {"method": "create_channel", "args": [{"name": "New"}]}, + {"method": "update_channel", "args": [-1, {"streams": [5]}]}, + {"method": "update_channel", "args": [-1, {"name": "Renamed"}]}, + ]) + + response, _ = await _commit(async_client, test_engine, upstream, payload) + + assert response.status_code == 424, response.text + detail = response.json()["detail"] + assert ("POST", "/api/channels/channels/") in upstream.calls + assert ("PATCH", "/api/channels/channels/101/") in upstream.calls + assert ("DELETE", "/api/channels/channels/101/") in upstream.calls # compensation hit the real id + assert 101 not in upstream.channels + assert detail["failed_index"] == 1 + assert detail["completed_writes"] == ["create_channel#0->101"] + assert detail["failed_write"] == "update_channel:101" + assert detail["not_applied"] == ["update_channel:101"] + assert detail["failed_outcome"] == "rejected" + assert detail["pre_mutation"] is False + assert detail["compensation_errors"] == [] + assert "-1" not in response.text + + execution, log_entry, evidence = _persisted(test_engine, detail["execution_id"]) + for persisted in (log_entry, evidence): + assert persisted["completed_targets"] == ["create_channel#0->101"] + assert persisted["failed_write"] == "update_channel:101" + assert persisted["not_applied"] == ["update_channel:101"] + assert "-1" not in json.dumps(log_entry) and "-1" not in json.dumps(evidence) + + +@pytest.mark.asyncio +async def test_compensation_failure_leaves_the_created_id_locatable_in_persisted_evidence( + async_client, test_engine, +): + upstream = Upstream() + upstream.script[("PATCH", "/api/channels/channels/101/")] = Upstream.reject(400) + upstream.script[("DELETE", "/api/channels/channels/101/")] = Upstream.reject(503) + payload = _payload([ + {"method": "create_channel", "args": [{"name": "New"}]}, + {"method": "update_channel", "args": [-1, {"streams": [5]}]}, + ]) + + response, _ = await _commit(async_client, test_engine, upstream, payload) + + assert response.status_code == 424, response.text + detail = response.json()["detail"] + assert 101 in upstream.channels # the orphan really is still there + assert detail["completed_writes"] == ["create_channel#0->101"] + assert detail["failed_write"] == "update_channel:101" + assert len(detail["compensation_errors"]) == 1 + assert detail["compensation_errors"][0].startswith("create_channel:") + assert detail["pre_mutation"] is False + + execution, log_entry, evidence = _persisted(test_engine, detail["execution_id"]) + assert execution.status == "failed" + for persisted in (log_entry, evidence): + assert persisted["completed_targets"] == ["create_channel#0->101"] + assert persisted["compensation_errors"] == detail["compensation_errors"] + + +# --------------------------------------------------------------------------- +# Item 3: a credential-bearing create payload never reaches the 424 body, +# the persisted evidence, or the backend log, while correlation survives. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_credential_bearing_logo_payload_is_absent_from_http_persistence_and_logs( + async_client, test_engine, caplog, +): + upstream = Upstream() + upstream.script[("POST", "/api/channels/logos/")] = Upstream.reject(429) + logo_url = f"http://provider.example/logo.png?token={CANARY}" + payload = _payload([ + {"method": "create_logo", "args": [{"name": "L", "url": logo_url}]}, + {"method": "create_channel", "args": [{"name": "New", "tvg_id": CANARY}]}, + {"method": "update_channel", "args": [-2, {"logo_id": -1}]}, + ]) + + caplog.set_level(logging.DEBUG) + with patch("dispatcharr_client._sleep", new=AsyncMock()): + response, _ = await _commit(async_client, test_engine, upstream, payload) + + assert response.status_code == 424, response.text + detail = response.json()["detail"] + # Correlation survives... + assert isinstance(detail["execution_id"], int) + assert detail["failed_write"] == "create_logo#0" + assert detail["not_applied"] == ["create_channel#1", "update_channel:pending(-2)"] + assert detail["failed_outcome"] == "rejected" + assert detail["pre_mutation"] is True + + execution, log_entry, evidence = _persisted(test_engine, detail["execution_id"]) + rendered = _everywhere(response, log_entry, evidence, execution) + # ...and the payload does not: not the token, not the URL, not the tvg_id. + assert CANARY not in rendered + assert "provider.example" not in rendered + assert "logo.png" not in rendered + # The backend log for the whole commit (client, replay, router) is clean too. + assert CANARY not in caplog.text + assert "provider.example" not in caplog.text + # The stored plan itself is the one place the payload legitimately lives + # (it is the replay program), and it predates this PR; the assertion here + # is on the DIAGNOSTIC fields the failure added, not on the plan blob. + assert log_entry["failed_write"] == "create_logo#0" + assert evidence["failed_write"] == "create_logo#0" diff --git a/frontend/package.json b/frontend/package.json index 78431b967..e993f7959 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -2,7 +2,7 @@ "name": "enhanced-channel-manager", "private": true, - "version": "0.18.2-0036", + "version": "0.18.2-0039", "type": "module", "scripts": { "dev": "vite", diff --git a/mcp-server/tests/test_gh1009_partial_failure_presentation.py b/mcp-server/tests/test_gh1009_partial_failure_presentation.py new file mode 100644 index 000000000..e17b40774 --- /dev/null +++ b/mcp-server/tests/test_gh1009_partial_failure_presentation.py @@ -0,0 +1,135 @@ +"""GH #1009 / PR #1010: how the sidecar presents a planned-commit 424. + +The backend answers a partial replay with ``424 Failed Dependency`` whose +``detail`` carries only bounded, payload-free descriptors +(``backend/tests/integration/test_gh1009_commit_partial_failure_evidence.py`` +proves that at the producer, including the credential canary). The sidecar +is a pass-through: ``ECMClient.post`` logs the first 500 bytes of the body +and ``_http_error`` surfaces ``detail`` in the raised message, which the +``run_channel_pipeline`` tool returns to the operator. These tests pin that +presentation: + +* the operator text and the sidecar log carry the correlation an operator + needs (``execution_id``, ``failed_write``, ``failed_outcome``, + ``not_applied``) verbatim; +* the sidecar adds nothing of its own — a control shows that a body which + DID carry a URL would reach the log and the tool text unchanged, which is + exactly why the redaction has to happen at the backend producer and is + asserted there. +""" +from __future__ import annotations + +import logging +from unittest.mock import patch + +import httpx +import pytest + +import ecm_client +from ecm_client import ECMClient, _http_error + +COMMIT_PATH = "/api/channel-pipeline/run/commit" +CANARY = "SECRET-TOKEN-8f3a9c-canary" + +# The exact field set the backend's commit handler emits on a partial replay +# (routers/channel_pipeline.py, ``except PartialReplayError``), with the +# descriptors the backend produced for a rejected credential-bearing +# ``create_logo`` first write. No payload contents appear by construction. +BACKEND_424_DETAIL = { + "message": "pipeline replay partially failed", + "execution_id": 91, + "failed_index": 0, + "failed_write": "create_logo#0", + "failed_outcome": "rejected", + "pre_mutation": True, + "completed_writes": [], + "not_applied": ["create_channel#1", "update_channel:pending(-2)"], + "compensation_errors": [], +} + + +def _sidecar_over(body: dict, status: int = 424) -> httpx.AsyncClient: + def handler(request: httpx.Request) -> httpx.Response: + assert request.method == "POST" and request.url.path == COMMIT_PATH + return httpx.Response(status, json=body, request=request) + + return httpx.AsyncClient(base_url="http://ecm", transport=httpx.MockTransport(handler)) + + +def _register_and_get_mcp(): + from tools.channel_pipeline import register + from mcp.server.fastmcp import FastMCP + + mcp = FastMCP("test") + register(mcp) + return mcp + + +def _text(result) -> str: + return result[0][0].text + + +@pytest.mark.asyncio +async def test_client_error_and_log_carry_the_safe_descriptors(caplog): + caplog.set_level(logging.DEBUG) + transport_client = _sidecar_over({"detail": BACKEND_424_DETAIL}) + try: + with patch.object(ecm_client, "_get_client", return_value=transport_client): + with pytest.raises(RuntimeError) as error: + await ECMClient().post(COMMIT_PATH, json_data={ + "plan_id": "p", "plan_hash": "h", "phase": "execute", + }) + finally: + await transport_client.aclose() + + message = str(error.value) + assert message.startswith(f"POST {COMMIT_PATH} -> HTTP 424") + for token in ("'execution_id': 91", "create_logo#0", "'failed_outcome': 'rejected'", + "create_channel#1", "update_channel:pending(-2)"): + assert token in message, message + assert "create_logo#0" in caplog.text and "424" in caplog.text + # Nothing that is not in the backend body can appear here. + assert CANARY not in message and CANARY not in caplog.text + assert "http://provider" not in message and "logo.png" not in caplog.text + + +@pytest.mark.asyncio +async def test_run_channel_pipeline_tool_returns_the_correlation_to_the_operator(caplog): + caplog.set_level(logging.DEBUG) + mcp = _register_and_get_mcp() + transport_client = _sidecar_over({"detail": BACKEND_424_DETAIL}) + try: + with patch.object(ecm_client, "_get_client", return_value=transport_client), \ + patch("tools.channel_pipeline.get_ecm_client", return_value=ECMClient()): + result = await mcp.call_tool( + "run_channel_pipeline", + {"dry_run": False, "plan_id": "p", "plan_hash": "h", "plan_phase": "execute"}, + ) + finally: + await transport_client.aclose() + + text = _text(result) + assert text.startswith("Error running auto-creation:") + assert "424" in text + assert "'execution_id': 91" in text + assert "create_logo#0" in text + assert "'failed_outcome': 'rejected'" in text + assert CANARY not in text and CANARY not in caplog.text + + +def test_http_error_renders_detail_verbatim_which_is_why_the_producer_must_be_safe(): + """Control: the sidecar does not redact. A body that carried a URL would + surface it unchanged, so the credential guarantee is the backend's and is + proven at that boundary, not here.""" + request = httpx.Request("POST", "http://ecm" + COMMIT_PATH) + leaky = {**BACKEND_424_DETAIL, "failed_write": f"create_logo:http://p.example/l.png?token={CANARY}"} + response = httpx.Response(424, json={"detail": leaky}, request=request) + exc = httpx.HTTPStatusError("424", request=request, response=response) + + rendered = str(_http_error("POST", COMMIT_PATH, exc)) + + assert CANARY in rendered # pass-through, by design: safety lives at the producer + safe = httpx.Response(424, json={"detail": BACKEND_424_DETAIL}, request=request) + assert CANARY not in str(_http_error( + "POST", COMMIT_PATH, httpx.HTTPStatusError("424", request=request, response=safe), + )) From af6f6904e6fae70ea7430860cf61ecda7877fc5e Mon Sep 17 00:00:00 2001 From: lukeeexd <31347888+lukeeexd@users.noreply.github.com> Date: Sun, 20 Sep 2026 20:05:45 +0100 Subject: [PATCH 4/4] fix(dispatcharr-client): one request sink in _request; isolate test credentials from the log redactor (#1009) Review round 3 on PR #1010, two current-head blockers: - CodeQL alert 2034 (py/partial-ssrf) fired on the 401 re-issue inside the new 429 loop, which duplicated the outbound request call. _request now has exactly one self._client.request call site: the 401 refresh sets a flag and loops back to it (at most one refresh per rate-limit attempt, the historical contract), and the 429 backoff increments the attempt counter and loops. Behaviour is unchanged for api-key mode (401 terminal), retry_on_401=False, and the retry budget; the auth, rate-limit, version-advisory and settings suites pin all of them. - DispatcharrClient.__init__ registers its credentials with the process-global log redactor, so the tests' one-character keys rewrote ordinary log text in a later test (seq***REDACTED***ence-1) and failed the persistent-log rotation test on CI. The new tests now use long, unique synthetic credentials, and tests/conftest.py gains an autouse fixture that snapshots the registered sensitive-value forms before each test and restores them afterwards, so no test can leak a registration into another. Co-Authored-By: Claude Fable 5.1 --- backend/dispatcharr_client.py | 28 +++++++++++++------ backend/tests/conftest.py | 25 +++++++++++++++++ ..._gh1009_commit_partial_failure_evidence.py | 8 +++++- .../test_dispatcharr_client_rate_limit.py | 4 +-- 4 files changed, 53 insertions(+), 12 deletions(-) diff --git a/backend/dispatcharr_client.py b/backend/dispatcharr_client.py index a3b162fb9..9713333ed 100644 --- a/backend/dispatcharr_client.py +++ b/backend/dispatcharr_client.py @@ -521,7 +521,15 @@ async def _request( try: waited = 0.0 - for attempt in range(RATE_LIMIT_MAX_RETRIES + 1): + attempt = 0 + # At most one token refresh per rate-limit attempt (the historical + # contract: a 401 triggers a refresh and ONE re-issue). + refreshed_this_attempt = False + # ONE request call site. The 401 re-issue and the 429 retry both + # loop back here instead of duplicating the outbound call, so the + # request URL has exactly one sink for static analysis and one + # place to reason about (PR #1010 review; CodeQL alert 2034). + while True: response = await self._client.request( method, f"{self.base_url}{path}", @@ -534,17 +542,17 @@ async def _request( # In api-key mode a 401 is terminal (the key is invalid or revoked), # and callers that opted out of the retry take the 401 as terminal # too rather than risk a rate-limited re-login (see the docstring). - if response.status_code == 401 and not self._uses_api_key and retry_on_401: + if ( + response.status_code == 401 + and not self._uses_api_key + and retry_on_401 + and not refreshed_this_attempt + ): logger.debug("[DISPATCHARR] Got 401, refreshing token and retrying: %s", method) await self._refresh_access_token() headers["Authorization"] = f"Bearer {self.access_token}" - response = await self._client.request( - method, - f"{self.base_url}{path}", - headers=headers, - timeout=request_timeout, - **kwargs, - ) + refreshed_this_attempt = True + continue # Rate limited: back off and retry (GH #1009). Exhausting the # budget raises an HTTPStatusError carrying the 429 so callers @@ -573,6 +581,8 @@ async def _request( ) waited += delay await _sleep(delay) + attempt += 1 + refreshed_this_attempt = False if response.status_code >= 400: logger.warning("[DISPATCHARR] API request failed: %s - status: %s", method, response.status_code) diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index e72edfbf4..e0b690d9f 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -33,6 +33,31 @@ assert models and export_models +@pytest.fixture(autouse=True) +def _isolate_registered_sensitive_values(): + """Keep the process-lifetime log redactor isolated between tests. + + ``DispatcharrClient.__init__`` (and other credential holders) register + their values with ``log_utils.register_sensitive_values``; production + never forgets them, by design. In the test process that means one test's + synthetic credential rewrites another test's ordinary log text (a one- + character key turned ``sequence-1`` into ``seq***REDACTED***ence-1`` and + broke the persistent-log rotation test on CI; PR #1010 / #1014 reviews). + Snapshot the registry before each test and restore it afterwards, so a + test only ever sees what was registered before it started. + """ + import log_utils + + with log_utils._sensitive_values_lock: + before = set(log_utils._sensitive_value_forms) + try: + yield + finally: + with log_utils._sensitive_values_lock: + log_utils._sensitive_value_forms.clear() + log_utils._sensitive_value_forms.update(before) + + def pytest_sessionfinish(session, exitstatus): """Clean up only the uniquely marked directory this process created.""" del session, exitstatus diff --git a/backend/tests/integration/test_gh1009_commit_partial_failure_evidence.py b/backend/tests/integration/test_gh1009_commit_partial_failure_evidence.py index 29a19f186..e5b786c70 100644 --- a/backend/tests/integration/test_gh1009_commit_partial_failure_evidence.py +++ b/backend/tests/integration/test_gh1009_commit_partial_failure_evidence.py @@ -98,7 +98,13 @@ def _reject(request: httpx.Request) -> httpx.Response: def _client(upstream: Upstream) -> DispatcharrClient: client = DispatcharrClient( - DispatcharrSettings(url="http://dispatcharr", auth_method="api_key", api_key="k") + DispatcharrSettings( + url="http://dispatcharr", auth_method="api_key", + # Long and unique: DispatcharrClient registers credentials with the + # process-global log redactor, and a short value would rewrite + # ordinary log text in later tests. + api_key="test-only-gh1009-api-key-7f3c9d2e1b", + ) ) client._client = httpx.AsyncClient(transport=httpx.MockTransport(upstream.handler)) return client diff --git a/backend/tests/unit/test_dispatcharr_client_rate_limit.py b/backend/tests/unit/test_dispatcharr_client_rate_limit.py index db4ca8717..d77bc144f 100644 --- a/backend/tests/unit/test_dispatcharr_client_rate_limit.py +++ b/backend/tests/unit/test_dispatcharr_client_rate_limit.py @@ -19,13 +19,13 @@ def _api_key_client() -> DispatcharrClient: return DispatcharrClient(DispatcharrSettings( - url="http://dispatcharr:8000", auth_method="api_key", dispatcharr_api_key="k", + url="http://dispatcharr:8000", auth_method="api_key", dispatcharr_api_key="test-only-rate-limit-api-key-4a8e2c6d", )) def _jwt_client() -> DispatcharrClient: return DispatcharrClient(DispatcharrSettings( - url="http://dispatcharr:8000", auth_method="password", username="u", password="p", + url="http://dispatcharr:8000", auth_method="password", username="test-only-rate-limit-user-91b7", password="test-only-rate-limit-pass-5d0f3a", ))