feat(render): park in-flight render jobs on Cloud Run shutdown (hardening 1/3) (#754)

## Summary

Addresses **hardening item 1 of 3** from
`docs/archive/2026-05-06-capacity-resilience-session.md`.

When Cloud Run autoscales an instance down (or rolls a deployment)
mid-render, the BackgroundTask was killed without writing failure state.
The job sat at \`rendering_video\` indefinitely until an operator
manually retried.

## Changes

- **`render_video_worker.py`**: registers/unregisters with
\`worker_registry\` (try/finally on the outer scope so it fires even on
unexpected exceptions). New \`park_active_render_jobs_for_shutdown()\`
helper transitions any in-flight render jobs to
\`RENDER_PENDING_CAPACITY\` (last_code: \`WORKER_SHUTDOWN\`) before
exit. Race-safe: skips jobs that have moved past \`RENDERING_VIDEO\`
between the registry snapshot and the per-job park.
- **`main.py`**: lifespan shutdown waits 480s (down from 600s) then runs
the park pass — 120s headroom for Firestore writes before SIGKILL. Cloud
Run gen2 default termination grace is 600s.
- **`docs/TROUBLESHOOTING.md`**: updated the \"Job stuck in
\`rendering_video\`\" runbook to describe the new self-recovery
behaviour.
- **`docs/archive/2026-05-06-capacity-resilience-session.md`**: marked
item 1 as shipped.
- **`pyproject.toml`**: 0.174.3 → 0.174.4

The auto-retry scheduler (\`/api/internal/retry-pending-render-jobs\`,
every 5 min via Cloud Scheduler) picks up parked jobs.

## Test plan

- [x] 6 new tests in \`test_render_video_worker_shutdown.py\`:
  - register on success, unregister on exception
- park transitions in-flight job to \`RENDER_PENDING_CAPACITY\` with
\`WORKER_SHUTDOWN\` code
  - race-safe skip when job moved past \`RENDERING_VIDEO\`
  - no-op when no render workers active
  - continues past per-job failures
- [x] Existing capacity-error tests
(\`test_render_video_worker_capacity.py\`) still pass —
register/unregister doesn't break the parking behaviour for actual
capacity errors.
- [ ] Prod verification: deploy → trigger a render → force a Cloud Run
revision rollover mid-render → confirm job lands in
\`render_pending_capacity\` and gets retried within 5 min. (Hard to
artificially induce SIGTERM without a real load event; will monitor next
time autoscaler scales down naturally.)

@coderabbitai ignore

Author: beveradb

backend/main.py

@@ -103,18 +103,37 @@ async def lifespan(app: FastAPI):
     yield
 
     # Shutdown - wait for any active workers to complete before terminating
-    # This prevents Cloud Run from killing workers mid-processing
+    # This prevents Cloud Run from killing workers mid-processing.
+    #
+    # Cloud Run gen2 default termination grace period is 600s. We wait up to
+    # 480s, leaving ~120s for the cleanup pass below to write Firestore state
+    # before SIGKILL. Without that headroom, render jobs that miss the wait
+    # would sit in `rendering_video` forever waiting for an operator.
     logger.info("Shutdown requested, checking for active workers...")
     if worker_registry.has_active_workers():
         active = worker_registry.get_active_workers()
         logger.info(f"Active workers found: {active}")
-        logger.info("Waiting for workers to complete (timeout: 600s)...")
-        completed = await worker_registry.wait_for_completion(timeout=600)  # 10 min max
+        logger.info("Waiting for workers to complete (timeout: 480s)...")
+        completed = await worker_registry.wait_for_completion(timeout=480)
         if not completed:
             logger.error(
                 "Shutdown timeout - some workers may not have completed cleanly. "
                 f"Remaining workers: {worker_registry.get_active_workers()}"
             )
+
+        # Park any still-active render jobs so the auto-retry scheduler can
+        # recover them. Safe to call unconditionally — it's a no-op if nothing
+        # render-related is in flight.
+        try:
+            from backend.workers.render_video_worker import park_active_render_jobs_for_shutdown
+            parked = park_active_render_jobs_for_shutdown()
+            if parked:
+                logger.warning(
+                    f"Parked {parked} in-flight render job(s) for auto-retry "
+                    f"due to shutdown"
+                )
+        except Exception as exc:
+            logger.error(f"Render-job shutdown park failed: {exc!r}")
     else:
         logger.info("No active workers, proceeding with shutdown")
 

backend/tests/test_render_video_worker_shutdown.py

@@ -0,0 +1,228 @@
+"""Tests for render-video worker shutdown handling.
+
+When Cloud Run autoscales an instance down (or rolls a deployment), it sends
+SIGTERM and waits up to 600s before SIGKILL. Without the changes covered
+here, an in-flight render would be killed silently and the job would sit at
+`rendering_video` indefinitely until an operator manually retried.
+
+The contract:
+  - process_render_video must register with worker_registry on entry and
+    unregister on exit (even on failure)
+  - park_active_render_jobs_for_shutdown must transition any in-flight render
+    job to RENDER_PENDING_CAPACITY so the auto-retry scheduler can recover it
+  - It must NOT touch jobs that have moved past RENDERING_VIDEO between the
+    registry snapshot and the park attempt (defends against the race where a
+    worker completes during shutdown)
+"""
+
+from unittest.mock import MagicMock, patch, AsyncMock
+
+import pytest
+
+from backend.models.job import JobStatus
+from backend.workers.registry import worker_registry
+
+
+def _build_minimal_job(status=JobStatus.RENDERING_VIDEO):
+    job = MagicMock()
+    job.artist = "Test Artist"
+    job.title = "Test Title"
+    job.input_media_gcs_path = "jobs/test/audio.flac"
+    job.style_assets = {}
+    job.style_params_gcs_path = None
+    job.subtitle_offset_ms = 0
+    job.prep_only = False
+    job.state_data = {}
+    job.file_urls = {}
+    job.status = status
+    return job
+
+
+@pytest.fixture(autouse=True)
+def reset_worker_registry():
+    """Each test starts with an empty registry — these tests touch shared state."""
+    worker_registry._active_workers.clear()
+    yield
+    worker_registry._active_workers.clear()
+
+
+@pytest.mark.asyncio
+async def test_worker_registers_and_unregisters_on_success():
+    """The render worker must register with worker_registry and unregister
+    on exit so the shutdown hook can wait for it.
+
+    Without registration, Cloud Run would kill the container the moment the
+    HTTP request that started the BackgroundTask returns — long before the
+    render finishes.
+    """
+    from backend.workers import render_video_worker as rvw
+
+    job = _build_minimal_job()
+
+    mock_job_manager = MagicMock()
+    mock_job_manager.get_job.return_value = job
+    mock_job_manager.transition_to_state.return_value = True
+
+    mock_encoding_service = MagicMock()
+    mock_encoding_service.is_enabled = True
+    # Return a successful render result
+    mock_encoding_service.render_video_on_gce = AsyncMock(return_value={
+        "output_files": [],
+        "metadata": {},
+    })
+
+    mock_storage = MagicMock()
+    mock_storage.file_exists.return_value = False
+
+    mock_worker_service = MagicMock()
+    mock_worker_service.trigger_video_worker = AsyncMock(return_value=True)
+
+    with patch.object(rvw, "JobManager", return_value=mock_job_manager), \
+         patch.object(rvw, "StorageService", return_value=mock_storage), \
+         patch.object(rvw, "get_settings"), \
+         patch.object(rvw, "create_job_logger", return_value=MagicMock()), \
+         patch.object(rvw, "setup_job_logging", return_value=MagicMock()), \
+         patch.object(rvw, "validate_worker_can_run", return_value=None), \
+         patch.object(rvw, "get_encoding_service", return_value=mock_encoding_service), \
+         patch("backend.services.worker_service.get_worker_service",
+               return_value=mock_worker_service):
+
+        result = await rvw.process_render_video("test-job-id")
+
+    assert result is True
+    # After completion, registry must be empty
+    assert worker_registry.get_active_workers() == {}
+
+
+@pytest.mark.asyncio
+async def test_worker_unregisters_even_on_unexpected_exception():
+    """A finally block must unregister even when the work itself blows up.
+
+    If unregister is skipped on errors, the registry leaks and shutdown park
+    would try (incorrectly) to park a job that has already failed.
+    """
+    from backend.workers import render_video_worker as rvw
+
+    mock_job_manager = MagicMock()
+    mock_job_manager.get_job.return_value = _build_minimal_job()
+
+    mock_encoding_service = MagicMock()
+    mock_encoding_service.is_enabled = True
+    mock_encoding_service.render_video_on_gce = AsyncMock(
+        side_effect=RuntimeError("unexpected boom")
+    )
+
+    with patch.object(rvw, "JobManager", return_value=mock_job_manager), \
+         patch.object(rvw, "StorageService"), \
+         patch.object(rvw, "get_settings"), \
+         patch.object(rvw, "create_job_logger", return_value=MagicMock()), \
+         patch.object(rvw, "setup_job_logging", return_value=MagicMock()), \
+         patch.object(rvw, "validate_worker_can_run", return_value=None), \
+         patch.object(rvw, "get_encoding_service", return_value=mock_encoding_service):
+
+        result = await rvw.process_render_video("test-job-id")
+
+    assert result is False
+    assert worker_registry.get_active_workers() == {}
+
+
+def test_park_active_render_jobs_parks_in_flight_render():
+    """park_active_render_jobs_for_shutdown must transition active render
+    jobs to RENDER_PENDING_CAPACITY with the WORKER_SHUTDOWN code marker."""
+    from backend.workers import render_video_worker as rvw
+
+    # Manually populate the registry to simulate an in-flight render
+    worker_registry._active_workers["job-A"] = {"render-video"}
+
+    mock_job_manager = MagicMock()
+    mock_job_manager.get_job.return_value = _build_minimal_job(JobStatus.RENDERING_VIDEO)
+    mock_job_manager.transition_to_state.return_value = True
+
+    with patch.object(rvw, "JobManager", return_value=mock_job_manager):
+        parked = rvw.park_active_render_jobs_for_shutdown()
+
+    assert parked == 1
+
+    # Should have written render_pending_capacity metadata with WORKER_SHUTDOWN code
+    state_calls = [
+        c for c in mock_job_manager.update_state_data.call_args_list
+        if c.args[1] == "render_pending_capacity"
+    ]
+    assert state_calls
+    pending_meta = state_calls[-1].args[2]
+    assert pending_meta["last_code"] == rvw.RENDER_WORKER_SHUTDOWN_CODE
+
+    # Should transition to RENDER_PENDING_CAPACITY
+    transitions = mock_job_manager.transition_to_state.call_args_list
+    assert any(
+        c.kwargs.get("new_status") == JobStatus.RENDER_PENDING_CAPACITY
+        for c in transitions
+    )
+
+
+def test_park_active_render_jobs_skips_completed_jobs():
+    """If the worker completed concurrently and the job moved past
+    RENDERING_VIDEO, the park must NOT clobber its terminal state.
+
+    Race: the worker_registry snapshot is taken at the start of the park;
+    between that and the per-job park call, the worker may finish and
+    transition the job to INSTRUMENTAL_SELECTED. Parking it back to
+    RENDER_PENDING_CAPACITY would re-trigger render and produce duplicate
+    work.
+    """
+    from backend.workers import render_video_worker as rvw
+
+    worker_registry._active_workers["job-already-done"] = {"render-video"}
+
+    mock_job_manager = MagicMock()
+    mock_job_manager.get_job.return_value = _build_minimal_job(
+        status=JobStatus.INSTRUMENTAL_SELECTED  # past render
+    )
+
+    with patch.object(rvw, "JobManager", return_value=mock_job_manager):
+        parked = rvw.park_active_render_jobs_for_shutdown()
+
+    assert parked == 0
+    mock_job_manager.update_state_data.assert_not_called()
+    mock_job_manager.transition_to_state.assert_not_called()
+
+
+def test_park_active_render_jobs_returns_zero_when_no_active_renders():
+    """No-op when nothing render-related is in flight (e.g. only audio jobs)."""
+    from backend.workers import render_video_worker as rvw
+
+    worker_registry._active_workers["job-X"] = {"audio", "lyrics"}
+
+    mock_job_manager = MagicMock()
+
+    with patch.object(rvw, "JobManager", return_value=mock_job_manager):
+        parked = rvw.park_active_render_jobs_for_shutdown()
+
+    assert parked == 0
+    # JobManager should not even be queried when there's nothing to do
+    mock_job_manager.get_job.assert_not_called()
+
+
+def test_park_active_render_jobs_continues_after_per_job_failure():
+    """One failing park must not block parking the rest — we have ~120s
+    before SIGKILL and we need to make a best effort."""
+    from backend.workers import render_video_worker as rvw
+
+    worker_registry._active_workers["job-fails"] = {"render-video"}
+    worker_registry._active_workers["job-ok"] = {"render-video"}
+
+    mock_job_manager = MagicMock()
+
+    def get_job_side_effect(job_id):
+        if job_id == "job-fails":
+            raise RuntimeError("Firestore unavailable")
+        return _build_minimal_job(JobStatus.RENDERING_VIDEO)
+
+    mock_job_manager.get_job.side_effect = get_job_side_effect
+    mock_job_manager.transition_to_state.return_value = True
+
+    with patch.object(rvw, "JobManager", return_value=mock_job_manager):
+        parked = rvw.park_active_render_jobs_for_shutdown()
+
+    # The successful job should still have been parked
+    assert parked == 1

backend/workers/render_video_worker.py

@@ -39,6 +39,7 @@
 from backend.services.storage_service import StorageService
 from backend.services.job_health_service import validate_worker_can_run
 from backend.config import get_settings
+from backend.workers.registry import worker_registry
 from backend.workers.worker_logging import create_job_logger, setup_job_logging, job_logging_context
 from backend.services.tracing import job_span, add_span_event, add_span_attribute
 from backend.services.encoding_service import get_encoding_service
@@ -47,6 +48,12 @@
     EncodingWorkerStartError,
 )
 
+
+# Sentinel exception code used when the render worker is parked because Cloud
+# Run is shutting the container down mid-render. Distinct from real GCE
+# capacity codes so log filters can tell them apart.
+RENDER_WORKER_SHUTDOWN_CODE = "WORKER_SHUTDOWN"
+
 # Import from lyrics_transcriber (submodule)
 from karaoke_gen.lyrics_transcriber.output.generator import OutputGenerator
 from karaoke_gen.lyrics_transcriber.output.countdown_processor import CountdownProcessor
@@ -91,12 +98,17 @@ async def process_render_video(job_id: str) -> bool:
 
     # Create job logger for remote debugging FIRST
     job_log = create_job_logger(job_id, "render_video")
-    
+
     # Log with structured markers for easy Cloud Logging queries
     logger.info(f"[job:{job_id}] WORKER_START worker=render-video")
     job_log.info("=== RENDER VIDEO WORKER STARTED ===")
     job_log.info(f"Job ID: {job_id}")
-    
+
+    # Register with worker registry so the FastAPI shutdown hook can wait for
+    # this worker AND, if Cloud Run forces termination, park the job in
+    # RENDER_PENDING_CAPACITY for the auto-retry scheduler to recover.
+    await worker_registry.register(job_id, "render-video")
+
     # Set up log capture for OutputGenerator
     log_handler = setup_job_logging(job_id, "render_video", *RENDER_VIDEO_WORKER_LOGGERS)
     job_log.info(f"Log handler attached for {len(RENDER_VIDEO_WORKER_LOGGERS)} loggers")
@@ -568,6 +580,11 @@ def progress_callback(progress: int):
         )
         job_manager.fail_job(job_id, f"Video render failed: {error_text}")
         return False
+    finally:
+        # Always unregister — if we don't, the shutdown hook will think this
+        # render is still in flight and try to park the (already-completed)
+        # job, which would produce a no-op transition + warning log spam.
+        await worker_registry.unregister(job_id, "render-video")
 
 
 def _park_job_for_capacity_retry(
@@ -629,5 +646,70 @@ def _extract_gcs_path(url: str) -> str:
     return url
 
 
+def park_active_render_jobs_for_shutdown() -> int:
+    """Park any in-flight render jobs in RENDER_PENDING_CAPACITY before exit.
+
+    Called by the FastAPI shutdown hook when Cloud Run is forcing the container
+    down (autoscaling, deploy, preemption) and one or more render workers
+    didn't finish in time. Without this, jobs sit at `rendering_video`
+    indefinitely until an operator manually retries.
+
+    Parking transitions the job to RENDER_PENDING_CAPACITY, which the
+    `/api/internal/retry-pending-render-jobs` Cloud Scheduler job picks up
+    every 5 minutes.
+
+    Only jobs currently in the RENDERING_VIDEO state are parked — if the
+    worker completed concurrently (or the job was already moved past this
+    stage by some other code path) we leave it alone.
+
+    Returns the number of jobs parked.
+    """
+    active = worker_registry.get_active_workers()
+    render_job_ids = [
+        job_id for job_id, worker_types in active.items()
+        if "render-video" in worker_types
+    ]
+    if not render_job_ids:
+        return 0
+
+    job_manager = JobManager()
+    parked = 0
+
+    for job_id in render_job_ids:
+        try:
+            job = job_manager.get_job(job_id)
+            if not job:
+                logger.warning(f"[job:{job_id}] Cannot park for shutdown: job not found")
+                continue
+            if job.status != JobStatus.RENDERING_VIDEO:
+                # Worker completed or job moved past this stage between the
+                # active-set snapshot and our park attempt — nothing to do.
+                logger.info(
+                    f"[job:{job_id}] Skipping shutdown park: status is "
+                    f"{job.status} (not RENDERING_VIDEO)"
+                )
+                continue
+
+            shutdown_marker = EncodingWorkerStartError(
+                "Render worker terminated by Cloud Run shutdown — auto-retry will resume",
+                code=RENDER_WORKER_SHUTDOWN_CODE,
+                vm_name="",
+                zone="",
+            )
+            _park_job_for_capacity_retry(job_manager, job_id, shutdown_marker)
+            parked += 1
+            logger.warning(
+                f"[job:{job_id}] WORKER_END worker=render-video status=shutdown_parked"
+            )
+        except Exception as exc:  # pragma: no cover - defensive
+            # Don't let one failed park block the others; we have a tight
+            # deadline before SIGKILL.
+            logger.error(
+                f"[job:{job_id}] Failed to park for shutdown: {exc!r}"
+            )
+
+    return parked
+
+
 # For compatibility with worker service
 render_video_worker = process_render_video