Encoding worker infra/auth failures: retry instead of terminal-failing the job

backend/services/encoding_errors.py

@@ -15,6 +15,44 @@
 })
 
 
+# Sentinel code stamped on an EncodingWorkerInfraError so log filters and the
+# park/retry path can tell a runtime worker-infra failure apart from a real
+# GCE VM-start failure.
+WORKER_INFRA_FAILURE_CODE = "worker_infra_failure"
+
+
+# Substrings (matched case-insensitively) that identify a worker-side *infrastructure*
+# failure — the VM booted and accepted the job but then could not talk to GCP to do
+# real work. The canonical case (job 6452888e, 2026-08-31) was a fallback VM that
+# could not reach the metadata server to fetch its default service-account token:
+#   "Failed to retrieve https://metadata.google.internal/.../service-accounts/default/
+#    ... Compute Engine Metadata server unavailable ... SSLCertVerificationError ..."
+# These are properties of the *VM*, not the encode job — the same work will succeed on
+# a healthy worker, so they must be retried/re-dispatched rather than failing the job.
+# Kept deliberately specific so a genuine encode error (bad codec, corrupt input) never
+# matches. Compared lowercase.
+WORKER_INFRA_ERROR_MARKERS: FrozenSet[str] = frozenset({
+    "metadata.google.internal",
+    "metadata server unavailable",
+    "compute engine metadata",
+    "service-accounts/default",
+    "sslcertverificationerror",
+    "certificate_verify_failed",
+    "could not automatically determine credentials",
+    "defaultcredentialserror",
+})
+
+
+def is_worker_infra_error(message: str) -> bool:
+    """True if a worker-reported error string looks like a VM-level infra/auth
+    failure (metadata server unreachable, SSL/cert, missing credentials) rather
+    than a genuine problem with the encode job itself."""
+    if not message:
+        return False
+    lowered = message.lower()
+    return any(marker in lowered for marker in WORKER_INFRA_ERROR_MARKERS)
+
+
 class EncodingWorkerStartError(Exception):
     """Raised when an attempt to start an encoding worker VM fails.
 
@@ -44,6 +82,20 @@ class EncodingWorkerCapacityError(EncodingWorkerStartError):
     """
 
 
+class EncodingWorkerInfraError(EncodingWorkerStartError):
+    """A worker VM accepted the job but then hit a VM-level infrastructure/auth
+    failure mid-run (e.g. it could not reach the GCE metadata server to fetch its
+    service-account token — job 6452888e, 2026-08-31).
+
+    Subclasses ``EncodingWorkerStartError`` on purpose: the failure is a property
+    of the *worker*, not the encode job, so it should flow through the same
+    recoverable path as a VM-start failure (the render worker parks the job for
+    auto-retry; the final-encode Cloud Run Job retries). Before raising this,
+    callers demote the offending VM so the retry lands on a healthy worker rather
+    than looping on the broken one.
+    """
+
+
 # Marker the encoding worker writes into a job's status when it was interrupted
 # by a worker-process restart (OOM, deploy, crash). Kept in sync with
 # gce_encoding/persistence.py `_RESTART_FAIL_CODE`.

backend/services/encoding_service.py

@@ -28,10 +28,13 @@
 from backend.config import get_settings
 from backend.services.encoding_errors import (
     ENCODING_RESTART_FAILURE_CODE,
+    WORKER_INFRA_FAILURE_CODE,
     EncodingJobLostError,
     EncodingJobNotFoundError,
     EncodingWorkerCapacityError,
+    EncodingWorkerInfraError,
     EncodingWorkerStartError,
+    is_worker_infra_error,
 )
 
 logger = logging.getLogger(__name__)
@@ -854,6 +857,42 @@ async def wait_for_completion(
                         f"Encoding job {job_id} was lost by the worker (restarted mid-run)",
                         job_id=job_id,
                     )
+                # A VM-level infra/auth failure (metadata server unreachable, SSL/cert,
+                # missing credentials) is a property of the *worker*, not this job — the
+                # same work succeeds on a healthy VM. Demote the broken worker and raise
+                # a recoverable signal so the render worker parks for auto-retry (and the
+                # final-encode Cloud Run Job retries) instead of failing the customer's
+                # job outright. See job 6452888e (2026-08-31). worker_url is the VM the
+                # job ran on; zone comes from active_override for logging.
+                if is_worker_infra_error(str(error)):
+                    logger.warning(
+                        f"[job:{job_id}] Worker reported an infrastructure/auth failure "
+                        f"(not a job problem) — demoting the worker and retrying: {error}"
+                    )
+                    failed_vm = ""
+                    if self._worker_manager:
+                        # Capture the offending VM before demoting (which clears the
+                        # active_override), then demote and force URL re-resolution so
+                        # the retry lands on a different worker.
+                        try:
+                            failed_vm = self._worker_manager.get_config().active_override_vm or ""
+                        except Exception:  # noqa: BLE001 — vm name is only for logging
+                            failed_vm = ""
+                        try:
+                            self._worker_manager.demote_active_worker(
+                                reason=f"job {job_id}: {error}"
+                            )
+                        except Exception as demote_err:  # noqa: BLE001
+                            logger.warning(
+                                f"[job:{job_id}] Could not demote worker after infra "
+                                f"failure (continuing to retry anyway): {demote_err}"
+                            )
+                        self._invalidate_cached_url()
+                    raise EncodingWorkerInfraError(
+                        f"Encoding worker infrastructure failure for job {job_id}: {error}",
+                        vm_name=failed_vm,
+                        code=WORKER_INFRA_FAILURE_CODE,
+                    )
                 raise RuntimeError(f"Encoding job {job_id} failed: {error}")
 
             await asyncio.sleep(poll_interval)

backend/services/encoding_worker_manager.py

@@ -586,6 +586,55 @@ def _clear_active_override(self) -> None:
         })
         logger.info("Cleared active_override (primary is healthy again)")
 
+    def demote_active_worker(self, reason: str = "") -> None:
+        """Demote the currently-serving fallback worker after a runtime infra failure.
+
+        Called when a worker VM accepted a job but then failed with a VM-level
+        infrastructure/auth error (e.g. it could not reach the metadata server —
+        job 6452888e). We only act when a *fallback* VM is serving (active_override
+        set): record a capacity-state cooldown for that VM's (machine_type, zone) so
+        ``ordered_candidates`` deprioritises its family for the cooldown window, then
+        clear the override so the retry re-resolves selection from the primary. When
+        the primary pair is serving we deliberately do nothing here — a metadata blip
+        on the stable c4d pair is vanishingly rare and family-demoting it would push
+        all traffic onto slow fallbacks; a bounded retry on the primary is safer.
+
+        Best-effort: never raises. Selection/telemetry hint only.
+        """
+        from backend.services.encoding_worker_preference import cooldown_key
+
+        try:
+            config = self.get_config()
+        except Exception as e:  # noqa: BLE001 — must never break the failure path
+            logger.warning("demote_active_worker: could not read config (%s): %s", reason, e)
+            return
+
+        vm = config.active_override_vm
+        zone = config.active_override_zone
+        if not vm:
+            # Primary pair was serving — see docstring; leave selection untouched.
+            logger.warning(
+                "Active encoding worker hit an infra failure while the primary was "
+                "serving; leaving selection unchanged for bounded retry: %s", reason,
+            )
+            return
+
+        key = cooldown_key({"vm_name": vm, "zone": zone})
+        if key:
+            try:
+                self._doc_ref().set(
+                    {"capacity_state": {key: datetime.now(UTC).isoformat()}}, merge=True
+                )
+                logger.warning(
+                    "Demoted fallback encoding worker %s (%s) after infra failure: %s",
+                    vm, key, reason,
+                )
+            except Exception as e:  # noqa: BLE001
+                logger.warning(
+                    "demote_active_worker: could not record cooldown for %s: %s", key, e
+                )
+        self._clear_active_override()
+
     # ------------------------------------------------------------------
     # Compute operation helpers
     # ------------------------------------------------------------------

backend/services/match_judge/service.py

@@ -106,8 +106,13 @@ async def judge_match(
         try:
             ai_verdict = await ai(artist, title, candidates, audio_tier)
         except Exception:
-            logger.exception(
-                "match-judge AI verification failed; keeping catalog verdict"
+            # Graceful degradation: a transient Vertex/Gemini blip just means we
+            # keep the catalog verdict. Log at WARNING (with traceback) so it does
+            # not page as a red "new error pattern" — it self-heals and needs no
+            # action. Downgraded 2026-08-31 after a single benign occurrence alerted.
+            logger.warning(
+                "match-judge AI verification failed; keeping catalog verdict",
+                exc_info=True,
             )
             return catalog_verdict
         if ai_verdict.confident and ai_verdict.kind != KIND_NONE:
@@ -122,7 +127,12 @@ async def judge_match(
     try:
         return await ai(artist, title, candidates, audio_tier)
     except Exception:
-        logger.exception("match-judge AI call failed; returning no-suggestion")
+        # Graceful degradation: returning no-suggestion is a safe fallback (the
+        # user just gets no auto-match). Log at WARNING (with traceback) so a
+        # transient AI blip does not page as a red "new error pattern".
+        logger.warning(
+            "match-judge AI call failed; returning no-suggestion", exc_info=True
+        )
         return MatchVerdict(
             KIND_NONE, True, artist, title, engine="ai", reason="ai failed"
         )

backend/tests/test_encoding_errors.py

@@ -0,0 +1,61 @@
+"""Tests for encoding worker error classification helpers."""
+
+from backend.services.encoding_errors import (
+    EncodingWorkerInfraError,
+    EncodingWorkerStartError,
+    is_worker_infra_error,
+)
+
+
+# The verbatim worker-reported error from the prod incident (job 6452888e).
+_METADATA_FAILURE = (
+    "Failed to retrieve https://metadata.google.internal/computeMetadata/v1/instance/"
+    "service-accounts/default/?recursive=true from the Google Compute Engine metadata "
+    "service. Compute Engine Metadata server unavailable. Last exception: "
+    "HTTPSConnectionPool(host='metadata.google.internal', port=443): Max retries "
+    "exceeded with url: /computeMetadata/v1/instance/service-accounts/default/"
+    "?recursive=true (Caused by SSLError(SSLCertVerificationError(1, '[SSL: "
+    "CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer "
+    "certificate (_ssl.c:1018)')))"
+)
+
+
+class TestIsWorkerInfraError:
+    def test_metadata_failure_is_infra(self):
+        assert is_worker_infra_error(_METADATA_FAILURE) is True
+
+    def test_matches_are_case_insensitive(self):
+        assert is_worker_infra_error(_METADATA_FAILURE.upper()) is True
+
+    def test_individual_markers(self):
+        for msg in (
+            "Compute Engine Metadata server unavailable",
+            "SSLCertVerificationError: certificate_verify_failed",
+            "google.auth.exceptions.DefaultCredentialsError: could not automatically "
+            "determine credentials",
+            "GET /computeMetadata/v1/instance/service-accounts/default/ failed",
+        ):
+            assert is_worker_infra_error(msg) is True, msg
+
+    def test_genuine_encode_error_is_not_infra(self):
+        # Real ffmpeg / input problems must stay terminal, never masquerade as infra.
+        for msg in (
+            "ffmpeg exploded",
+            "Invalid data found when processing input",
+            "Unknown encoder 'libx265'",
+            "Output file is empty, nothing was encoded",
+            "",
+        ):
+            assert is_worker_infra_error(msg) is False, msg
+
+    def test_none_is_not_infra(self):
+        assert is_worker_infra_error(None) is False  # type: ignore[arg-type]
+
+
+class TestEncodingWorkerInfraError:
+    def test_is_recoverable_start_error_subclass(self):
+        """Subclassing EncodingWorkerStartError is what routes it through the render
+        worker's park-for-auto-retry path (and the final-encode Cloud Run Job retry)."""
+        err = EncodingWorkerInfraError("boom", vm_name="encoding-worker-fallback-c4a")
+        assert isinstance(err, EncodingWorkerStartError)
+        assert err.vm_name == "encoding-worker-fallback-c4a"

backend/tests/test_encoding_service.py

@@ -746,6 +746,58 @@ async def mock_get_status(job_id, worker_url=None):
             except RuntimeError:
                 pass
 
+    @pytest.mark.asyncio
+    async def test_worker_infra_failure_raises_infra_error_and_demotes(self, encoding_service):
+        """A metadata/auth failure reported by the worker is a VM problem, not a job
+        problem (job 6452888e): raise the recoverable EncodingWorkerInfraError, demote
+        the broken worker and invalidate the URL cache so the retry lands elsewhere."""
+        from backend.services.encoding_errors import EncodingWorkerInfraError
+
+        metadata_error = (
+            "Failed to retrieve https://metadata.google.internal/computeMetadata/v1/"
+            "instance/service-accounts/default/?recursive=true from the Google Compute "
+            "Engine metadata service. Compute Engine Metadata server unavailable. Last "
+            "exception: SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_"
+            "FAILED] certificate verify failed: unable to get local issuer certificate'))"
+        )
+
+        async def mock_get_status(job_id, worker_url=None):
+            return {"status": "failed", "error": metadata_error}
+
+        mock_manager = MagicMock()
+        mock_manager.get_config.return_value.active_override_vm = "encoding-worker-fallback-c4a"
+        encoding_service._worker_manager = mock_manager
+
+        with patch.object(encoding_service, "get_job_status", side_effect=mock_get_status), \
+             patch.object(encoding_service, "_invalidate_cached_url") as mock_invalidate, \
+             patch("asyncio.sleep", new_callable=AsyncMock), \
+             patch("asyncio.get_event_loop") as mock_loop:
+            mock_loop.return_value.time.return_value = 0
+            with pytest.raises(EncodingWorkerInfraError) as exc_info:
+                await encoding_service.wait_for_completion("j1")
+
+        # Captured the offending VM, demoted it, and forced URL re-resolution.
+        assert exc_info.value.vm_name == "encoding-worker-fallback-c4a"
+        mock_manager.demote_active_worker.assert_called_once()
+        mock_invalidate.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_worker_infra_failure_without_manager_still_recoverable(self, encoding_service):
+        """Even with no worker manager wired (static-URL fallback), an infra failure is
+        classified as recoverable (EncodingWorkerInfraError), never a terminal error."""
+        from backend.services.encoding_errors import EncodingWorkerInfraError
+
+        async def mock_get_status(job_id, worker_url=None):
+            return {"status": "failed", "error": "Compute Engine Metadata server unavailable"}
+
+        encoding_service._worker_manager = None
+        with patch.object(encoding_service, "get_job_status", side_effect=mock_get_status), \
+             patch("asyncio.sleep", new_callable=AsyncMock), \
+             patch("asyncio.get_event_loop") as mock_loop:
+            mock_loop.return_value.time.return_value = 0
+            with pytest.raises(EncodingWorkerInfraError):
+                await encoding_service.wait_for_completion("j1")
+
     @pytest.mark.asyncio
     async def test_pending_does_not_count_toward_run_timeout(self, encoding_service):
         """Time spent queued (pending) is bounded by queue_timeout, not the per-run

backend/tests/test_encoding_worker_manager.py

@@ -1072,3 +1072,67 @@ def test_legacy_candidates_without_flag_use_position(self, manager, mock_db, moc
 
         assert result["fell_back"] is False
         assert result["vm_name"] == "encoding-worker-blue"
+
+
+# ---------------------------------------------------------------------------
+# demote_active_worker: skip a broken fallback VM after a runtime infra failure
+# ---------------------------------------------------------------------------
+
+class TestDemoteActiveWorker:
+    """A worker VM that accepted a job but then hit a VM-level infra/auth failure
+    (e.g. could not reach the metadata server — job 6452888e) must be demoted so
+    the retry lands on a healthy worker."""
+
+    def test_demotes_fallback_and_clears_override(self, manager):
+        """When a fallback is serving, record its family cooldown then clear the
+        override so selection re-resolves from the primary."""
+        doc_ref = manager._doc_ref()
+        doc_ref.get.return_value.exists = True
+        doc_ref.get.return_value.to_dict.return_value = _make_firestore_data(
+            active_override_vm="encoding-worker-fallback-c4a",
+            active_override_ip="10.128.0.99",
+            active_override_zone="us-central1-c",
+            active_override_set_at="2026-08-31T12:00:00Z",
+        )
+
+        manager.demote_active_worker(reason="job 6452888e: metadata server unavailable")
+
+        # Recorded a cooldown for the fallback's (machine_type@zone).
+        assert doc_ref.set.called
+        set_args, set_kwargs = doc_ref.set.call_args
+        assert "capacity_state" in set_args[0]
+        assert "c4-highcpu-32@us-central1-c" in set_args[0]["capacity_state"]
+        assert set_kwargs.get("merge") is True
+
+        # Cleared the active_override so the retry starts from the primary.
+        override_clears = [
+            c for c in doc_ref.update.call_args_list
+            if c.args and "active_override_vm" in c.args[0]
+        ]
+        assert override_clears, "expected active_override to be cleared"
+        assert override_clears[-1].args[0]["active_override_vm"] is None
+
+    def test_no_op_when_primary_is_serving(self, manager):
+        """When no fallback override is set (primary pair serving), demote is a no-op:
+        we don't family-demote the fast c4d primary on a rare blip."""
+        doc_ref = manager._doc_ref()
+        doc_ref.get.return_value.exists = True
+        doc_ref.get.return_value.to_dict.return_value = _make_firestore_data()  # no override
+
+        manager.demote_active_worker(reason="job zzz: metadata server unavailable")
+
+        assert not doc_ref.set.called
+        override_clears = [
+            c for c in doc_ref.update.call_args_list
+            if c.args and "active_override_vm" in c.args[0]
+        ]
+        assert not override_clears
+
+    def test_never_raises_on_config_read_failure(self, manager):
+        """Demotion is best-effort telemetry — a Firestore hiccup must not break the
+        failure path it runs on."""
+        doc_ref = manager._doc_ref()
+        doc_ref.get.side_effect = RuntimeError("firestore unavailable")
+
+        # Should swallow the error rather than propagate.
+        manager.demote_active_worker(reason="job qqq")