v0.18.2-0043: report partial commit failures and bound 429 retries (#1009)

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.

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:

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",