refactor(native-ingest): shared exception trio + base _failure_result (#899 DRY 2/3)

DRY pass commit 2. The two adapters re-implemented the same auth→drift→429→404
→http→network mapping in _failure_result; only the exception classes + drift
phrasing differed (divergence-bug risk: a new error_type handled in one and not
the other).

- native_ingest_common gains NativeIngestError / NativeAuthError / NativeDriftError
  (status_code + retry_after on the base). Patreon{API,Auth,Drift}Error and
  SubscribeStar{API,Auth,Drift}Error now subclass them via multiple inheritance,
  keeping their isinstance-distinct platform names.
- Ingester._failure_result (base) does the whole mapping via the shared
  NativeAuthError/NativeDriftError taxonomy + status_code; a new platform gets it
  free. New drift_label kwarg supplies the per-platform API_DRIFT phrasing
  ("Patreon API" / "SubscribeStar markup"), preserving the existing message
  (test asserts "Patreon API changed").
- Both adapters drop their near-identical _failure_result overrides and their now
  -unused DownloadResult/ErrorType/*Auth/*Drift imports.

Verified at every consumer (rule 93/§8b): test_patreon_ingester (auth/drift/429/
404/network) and test_subscribestar_native (_failure_result mapping) both exercise
the base method now. Remaining: ingest_core L1/L3 logging (3/3).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/app/services/ingest_core.py

@@ -36,6 +36,7 @@ from sqlalchemy import delete, func, select, text
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from .gallery_dl import DownloadResult, ErrorType, make_run_stats
+from .native_ingest_common import NativeAuthError, NativeDriftError
 
 log = logging.getLogger(__name__)
 
@@ -88,6 +89,7 @@ class Ingester:
         ledger_key: Callable[[object], str],
         platform: str,
         error_base: type[Exception],
+        drift_label: str | None = None,
     ):
         self.client = client
         self.downloader = downloader
@@ -99,6 +101,10 @@ class Ingester:
         self._ledger_key = ledger_key
         self._platform = platform
         self._error_base = error_base
+        # Human label for the API_DRIFT message ("<label> changed — ingester needs
+        # update"). Defaults to the platform name; adapters pass a richer phrase
+        # (e.g. "Patreon API", "SubscribeStar markup").
+        self._drift_label = drift_label or platform
 
     # -- public ------------------------------------------------------------
 
@@ -601,13 +607,41 @@ class Ingester:
     # -- failure mapping (adapter overrides) -------------------------------
 
     def _failure_result(self, exc: Exception, _result) -> DownloadResult:
-        """Map a platform client-error to a typed failed DownloadResult. The base
-        gives a safe default; adapters override with their exception taxonomy."""
-        log.warning("%s ingest failed: %s", self._platform, exc)
-        return _result(
+        """Map a platform client-error to a loud, typed failed DownloadResult —
+        NEVER a silent zero-download "success". The mapping is shared across
+        platforms via the NativeAuthError/NativeDriftError taxonomy (the platform
+        client raises subclasses), so a new platform gets it for free:
+          - NativeAuthError  → AUTH_ERROR   (rotate the credential)
+          - NativeDriftError → API_DRIFT    (the ingester/scraper needs updating)
+          - HTTP 429 / 404   → RATE_LIMITED / NOT_FOUND
+          - other HTTP status→ HTTP_ERROR;  transport failure → NETWORK_ERROR
+        Auth/Drift are matched first (they also carry a status_code in some paths).
+        """
+        message = str(exc)
+        if isinstance(exc, NativeAuthError):
+            error_type = ErrorType.AUTH_ERROR
+        elif isinstance(exc, NativeDriftError):
+            error_type = ErrorType.API_DRIFT
+            message = f"{self._drift_label} changed — ingester needs update: {message}"
+        else:
+            status = getattr(exc, "status_code", None)
+            if status == 429:
+                error_type = ErrorType.RATE_LIMITED
+            elif status == 404:
+                error_type = ErrorType.NOT_FOUND
+            elif status is not None:
+                error_type = ErrorType.HTTP_ERROR
+            else:
+                error_type = ErrorType.NETWORK_ERROR
+        log.warning("%s ingest failed (%s): %s", self._platform, error_type.value, message)
+        result = _result(
             success=False, return_code=1,
-            error_type=ErrorType.UNKNOWN_ERROR, error_message=str(exc),
+            error_type=error_type, error_message=message,
         )
+        # plan #708 B1: carry the server's Retry-After up to the cooldown.
+        if error_type == ErrorType.RATE_LIMITED:
+            result.retry_after_seconds = getattr(exc, "retry_after", None)
+        return result
 
     # -- seen-ledger (short-lived sessions) --------------------------------