diff --git a/CHANGELOG.md b/CHANGELOG.md index 44195ff56..201c1d947 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-0043).** `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. - **A provider's daily slot rollover no longer queues the whole group as duplicates (GitHub #1015, build 0.18.2-0042).** Event providers reuse a fixed pool of slot names and roll the fixture and the airing over every day, so yesterday's channel for a slot and today's stream for the same slot share every template word. The M3U-refresh dedup scorer scored them at 0.86-0.99 — above the default 0.80 — queued the stream for operator review, and deferred the channel creation, leaving the group on yesterday's fixtures until orphan cleanup removed them. A stream whose name states a different valid airing from the candidate's is now recognised as a rollover and created instead of queued; only real month names, days, clock times and supported zone labels (bare or with an offset) form an airing, a false prefix cannot hide a real trailing one, and a quality or provider suffix after the clock is not a zone, so ordinary names and genuine duplicates score as before. A channel a stream is already attached to is never offered as a merge candidate for that stream, and membership is inspected only on channels the group and name filters already admit. A run that created nothing because streams were deferred now reports the deferred streams, the deferred create actions when they differ, and the distinct pending-merge rows it is waiting on, in the run log, the post-refresh summary and the single warning an unattended failed-action run emits — the task engine's *Completed with Warnings* notification, or the task's own combined *Capped, with errors* notification when the run also hit the created-channel cap; a deferred stream's follow-on EPG step names the queue instead of failing with a bare "No channel context for assign_epg", and a later create attempt for the same stream supersedes that explanation. Dry-run previews do not yet predict these deferrals; that is tracked separately in [#1018](https://github.com/MotWakorb/enhancedchannelmanager/issues/1018) and is not part of this fix. - **A duplicate logo URL no longer aborts a planned channel-pipeline commit (GitHub #1013; build 0.18.2-0041).** Dispatcharr rejects `POST /api/channels/logos/` with 400 when a logo row with the same URL already exists, and logo rows outlive the channels that used them, so a stale row from an earlier event cycle made the fail-fast replay raise `PartialReplayError` at write 0 and the commit return 502 with none of the planned channel creates, stream merges or EPG assignments landed. The executor now resolves a stream's logo URL to an existing Dispatcharr logo before planning a create, so the plan carries no `create_logo` write for a known logo and the created channel references the existing row. The client's `create_logo` is idempotent on URL: it resolves an existing row before the POST (callers that have already established the URL is absent, such as the bulk channel create with its per-run catalog index, skip that pre-check so the catalog is scanned once per run, not once per URL) and again after a 400 for a row created in between. A reused row is returned marked as reused, so the DBAS logo importer maps it for channel reattach but neither ledgers it for compensating delete (a rollback would have deleted a logo that pre-dates the restore) nor counts it as created. On failure the client logs and raises only the status and a fixed classification (`duplicate_url`, `validation`, `server_error`, `error`), never the upstream response body, which can echo a credentialed logo URL. In the replay, `create_logo` is a soft-fail write classified by recorded write method, not HTTP status: a `create_logo` that still fails is skipped, a later payload field that referenced its temporary id is omitted from the request rather than sent as `null` (so an existing channel keeps its current logo and a new channel is created without one), an update left with no fields is not sent at all, and a positional dependency on the skipped create still aborts. Channel creates, stream merges, EPG writes and every other structural write keep the fail-fast abort and compensation, and a partial failure after a skipped write reports the failing write's true plan position. The execution record carries a `replay_write_skipped` warning naming each skipped write and the channel it would have touched, and the journal records only mutations that actually happened, so an operator recognises cosmetic degradation from the execution's warnings after the useful channel work has completed. - **Assign EPG against a dummy EPG source no longer fails for channels the same run just created (GitHub #1011; build 0.18.2-0040).** A standard `create_channel` → `assign_epg` rule targeting an ECM dummy EPG source failed every assignment for the channels it had just created: the source's XMLTV still described the channels as they were before the run, and the executor only deferred when the source had no entries at all. In a direct run, a no-match against a recognised dummy source now defers into the existing Pass 5 regenerate, refresh and retry when the channel was created in this run (tracked by channel id, so same-named channels in different groups are both eligible), so one run assigns the guide data, and changes the channel's TVG-ID only when `set_tvg_id` is enabled. The deferred retry is bound to the channel it was queued for, is attributed to the originating rule in the per-rule outcome if it fails, and is terminal: a retry that still finds no entry is a failed action with an actionable reason, never a re-deferral. Each dummy EPG profile gains only the groups whose deferred assignments target its own feed. A planned run (prepare/commit) cannot regenerate the feed before committing, so there the assignment fails with a reason that says so rather than being recorded as fulfilled. Pre-existing channels keep the no-match-is-failure behaviour, and their error now says the dummy source has no entry for the channel yet instead of a generic no-match. The run-level failure summary carries each sampled action's reason alongside the rule and action type; reasons are redacted of credential-bearing content and bounded in size. diff --git a/backend/dispatcharr_client.py b/backend/dispatcharr_client.py index fe0742522..f48925a26 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 @@ -253,6 +257,94 @@ 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 +# 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, 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 + 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. + """ + detail = reason or f"after {RATE_LIMIT_MAX_RETRIES} retries" + return httpx.HTTPStatusError( + f"Dispatcharr rate limited (429) {detail}", + request=getattr(response, "request", None), + response=response, + ) + + +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 + + # Marker set on a logo row that ``create_logo`` RESOLVED instead of created # (PR #1014 review item 1). Absent on a freshly created row. LOGO_REUSED_KEY = "ecm_reused" @@ -353,13 +445,34 @@ 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, - }, - ) + waited = 0.0 + 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, 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() self.access_token = data["access"] @@ -455,22 +568,16 @@ 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}" + waited = 0.0 + 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}", @@ -479,6 +586,52 @@ 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 + 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}" + refreshed_this_attempt = True + continue + + # 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, 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) + attempt += 1 + refreshed_this_attempt = False + if response.status_code >= 400: logger.warning("[DISPATCHARR] API request failed: %s - status: %s", method, response.status_code) else: diff --git a/backend/main.py b/backend/main.py index bb9ea0274..01e36aaec 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-0042", + version="0.18.2-0043", 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 a9a89d58e..1efaf871c 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-0042" +APP_VERSION = "0.18.2-0043" # 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/routers/channel_pipeline.py b/backend/routers/channel_pipeline.py index edb466b92..318b1a254 100644 --- a/backend/routers/channel_pipeline.py +++ b/backend/routers/channel_pipeline.py @@ -2070,15 +2070,35 @@ 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, + "failed_outcome": exc.failed_outcome, "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. + # ``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=502, + status_code=424, detail={ "message": "pipeline replay partially failed", + "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, "compensation_errors": exc.compensation_errors, }, ) from exc diff --git a/backend/services/pipeline_write_plan.py b/backend/services/pipeline_write_plan.py index bdbfa8646..7c77d76c7 100644 --- a/backend/services/pipeline_write_plan.py +++ b/backend/services/pipeline_write_plan.py @@ -6,6 +6,8 @@ from dataclasses import dataclass, field from typing import Any +import httpx + from services.mutation_plan_store import canonical_hash logger = logging.getLogger(__name__) @@ -87,6 +89,10 @@ 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, ...) + # Writes whose failure must not abort a replay. A logo is cosmetic: losing it # costs one image, while aborting costs every planned channel create, stream # merge and EPG assignment behind it. Dispatcharr answers 400 for a duplicate @@ -114,13 +120,97 @@ class ReplayOutcome: class PartialReplayError(RuntimeError): - """Upstream has no transaction; exposes exactly how far replay reached.""" + """Upstream has no transaction; exposes exactly how far replay reached. + + 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]): + 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, failed_outcome: str = FAILED_OUTCOME_UNKNOWN, + ): 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 + 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. + + 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: + """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: @@ -345,11 +435,15 @@ def _skip(index: int, write: PlannedWrite, reason: str, error_type: str | None = results.append(None) next_temp = -1 - completed: list[tuple[PlannedWrite, list[Any], Any]] = [] + completed: list[tuple[int, 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 failed_index = 0 try: for index, write in enumerate(plan.writes): failed_index = index + attempted = False args = resolve(write.args) kwargs = resolve(write.kwargs) if ( @@ -365,6 +459,7 @@ def _skip(index: int, write: PlannedWrite, reason: str, error_type: str | None = ) _skip(index, write, "dependency_skipped") continue + attempted = True if write.event_sync: from services.event_sync_cleanup import apply_change result = await apply_change(client, write.event_sync, execution_id) @@ -392,7 +487,7 @@ def _skip(index: int, write: PlannedWrite, reason: str, error_type: str | None = remap[next_temp] = int(real_id) next_temp -= 1 results.append(result) - completed.append((write, args, result)) + completed.append((index, write, args, result)) except Exception as exc: compensation_errors: list[str] = [] if any(write.event_sync for write in plan.writes): @@ -402,7 +497,7 @@ def _skip(index: int, write: PlannedWrite, reason: str, error_type: str | None = compensation_errors.append(recovery["error"]) # Best effort for reversible writes. Deletes and profile membership are # explicitly not recreated because upstream cannot preserve their IDs. - for done, args, result in reversed(completed): + for _, done, args, result in reversed(completed): if done.event_sync: continue try: @@ -424,11 +519,34 @@ def _skip(index: int, write: PlannedWrite, reason: str, error_type: str | None = 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 ''}" + describe_write_target( + item[1], item[0], remap, + created_id=( + item[3].get("id") if item[1].method.startswith("create_") + and isinstance(item[3], dict) else None + ), + ) for item in 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_index, completed_targets, compensation_errors, + failed_write=failed_write, + not_applied=not_applied, + 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/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..80d179bcc --- /dev/null +++ b/backend/tests/integration/test_gh1009_commit_partial_failure_evidence.py @@ -0,0 +1,355 @@ +"""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", + # 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 + + +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_create_payload_is_absent_from_http_persistence_and_logs( + async_client, test_engine, caplog, +): + upstream = Upstream() + upstream.script[("POST", "/api/channels/channels/")] = Upstream.reject(429) + logo_url = f"http://provider.example/logo.png?token={CANARY}" + payload = _payload([ + {"method": "create_channel", "args": [{"name": "New", "logo_url": logo_url, "tvg_id": CANARY}]}, + {"method": "update_channel", "args": [-1, {"name": "Later"}]}, + ]) + + 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_channel#0" + assert detail["not_applied"] == ["update_channel:pending(-1)"] + 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_channel#0" + assert evidence["failed_write"] == "create_channel#0" diff --git a/backend/tests/routers/test_channel_pipeline.py b/backend/tests/routers/test_channel_pipeline.py index ecf6e7a40..2079fc291 100644 --- a/backend/tests/routers/test_channel_pipeline.py +++ b/backend/tests/routers/test_channel_pipeline.py @@ -5394,3 +5394,75 @@ 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=["delete_channel:8"], pre_mutation=True, + failed_outcome="rejected", + ) + 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["failed_outcome"] == "rejected" + assert detail["completed_writes"] == [] + # 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_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 2ad8611ea..dcdfb9d82 100644 --- a/backend/tests/services/test_pipeline_write_plan.py +++ b/backend/tests/services/test_pipeline_write_plan.py @@ -2,6 +2,7 @@ from pathlib import Path from unittest.mock import AsyncMock +import httpx import pytest from services.pipeline_write_plan import ( @@ -106,6 +107,26 @@ async def test_drift_rejects_before_any_replay_write(): 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], {}), + ], + ) + + # --------------------------------------------------------------------------- # Logo writes are cosmetic: a failed create_logo must not abort the replay. # Dispatcharr answers 400 for a duplicate logo URL and logo rows outlive their @@ -124,6 +145,150 @@ def _logo_then_channel_plan(): ) +@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.failed_outcome == "rejected" + assert exc.completed == [] + assert exc.not_applied == ["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.failed_outcome == "rejected" + assert exc.completed == ["update_channel:7"] + assert exc.not_applied == [] + 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 + 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_replay_continues_past_a_failed_logo_create_and_lands_the_channel(): live = AsyncMock() @@ -142,6 +307,56 @@ async def test_replay_continues_past_a_failed_logo_create_and_lands_the_channel( ] +@pytest.mark.asyncio +async def test_credential_bearing_create_payload_never_reaches_diagnostics(): + """Item 3: a create payload may carry a provider token; diagnostics may not.""" + token = "SECRET-TOKEN-8f3a9c" + live = AsyncMock() + live.create_channel.side_effect = _rate_limited() + plan = PipelineWritePlan(writes=[ + PlannedWrite("create_channel", [{"name": "New", "tvg_id": token}], {}), + PlannedWrite("update_channel", [-1, {"name": "Later"}], {}), + ]) + with pytest.raises(PartialReplayError) as error: + await replay_write_plan(live, plan) + exc = error.value + assert exc.failed_write == "create_channel#0" + assert exc.not_applied == ["update_channel:pending(-1)"] + 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() + + @pytest.mark.asyncio async def test_replay_still_aborts_and_compensates_when_a_channel_create_fails(): live = AsyncMock() @@ -150,7 +365,7 @@ async def test_replay_still_aborts_and_compensates_when_a_channel_create_fails() with pytest.raises(PartialReplayError) as info: await replay_write_plan(live, _logo_then_channel_plan()) assert info.value.failed_index == 1 - assert info.value.completed == ["create_logo:{'name': 'Snooker', 'url': 'http://l/x.png'}"] + assert info.value.completed == ["create_logo#0->765"] live.update_channel.assert_not_awaited() @@ -308,7 +523,7 @@ async def test_failed_index_is_the_true_plan_position_after_soft_skips(): with pytest.raises(PartialReplayError) as info: await replay_write_plan(live, plan) assert info.value.failed_index == 2 - assert info.value.completed == ["create_channel:{'name': 'New', 'streams': [5]}"] + assert info.value.completed == ["create_channel#1->101"] live.delete_channel.assert_awaited_once_with(101) # compensation unchanged 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..d77bc144f --- /dev/null +++ b/backend/tests/unit/test_dispatcharr_client_rate_limit.py @@ -0,0 +1,305 @@ +"""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="test-only-rate-limit-api-key-4a8e2c6d", + )) + + +def _jwt_client() -> DispatcharrClient: + return DispatcharrClient(DispatcharrSettings( + url="http://dispatcharr:8000", auth_method="password", username="test-only-rate-limit-user-91b7", password="test-only-rate-limit-pass-5d0f3a", + )) + + +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 + + +# --------------------------------------------------------------------------- +# 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 ab70d26e7..6682dfeeb 100644 --- a/docs/user_guide/integrations/mcp.md +++ b/docs/user_guide/integrations/mcp.md @@ -740,7 +740,33 @@ 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 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 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. diff --git a/frontend/package.json b/frontend/package.json index 690853ede..afa2c324a 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -2,7 +2,7 @@ "name": "enhanced-channel-manager", "private": true, - "version": "0.18.2-0042", + "version": "0.18.2-0043", "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), + ))