fix: resilient processed-video upload (resumable + retry)

backend/pipeline/storage.py

@@ -1,9 +1,24 @@
 from supabase import create_client
+import base64
 import os
 import subprocess
+import time
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
+import httpx
+
+# The processed video is uploaded via Supabase's resumable (TUS) endpoint rather
+# than the standard /object endpoint, which rejects bodies over ~50MB (413) and is
+# subject to storage3's 20s-per-request timeout — a larger clip under load hits one
+# or the other and the whole job is marked failed. TUS sends 6MB chunks (each well
+# under both limits). Uploads also gained retry, which the download path already had.
+_TUS_CHUNK_SIZE = 6 * 1024 * 1024  # Supabase requires 6MB TUS chunks (last may be smaller)
+_UPLOAD_MAX_ATTEMPTS = 4
+_STORAGE_TIMEOUT_SEC = 120.0  # per-request timeout; also capped by the overall upload budget
+_UPLOAD_BUDGET_SEC = 600.0  # overall wall-clock per artifact across retries (< the 1800s job cap)
+_FFMPEG_BUDGET_SEC = 900  # overall encode budget shared across codec attempts (< the 1800s job cap)
+
 def make_streamable_mp4(input_path: str, source_audio_path: str | None = None) -> str:
     """
     Re-encode and remux to a browser-streamable MP4 (moov atom first).
@@ -20,8 +35,15 @@ def make_streamable_mp4(input_path: str, source_audio_path: str | None = None) -
     """
     input_path = Path(input_path)
     output_path = input_path.with_suffix("").with_name(input_path.stem + "_web.mp4")
+    # One deadline shared across both codec attempts: a hung nvenc followed by a
+    # hung libx264 could otherwise each burn the full budget and take the whole job.
+    encode_deadline = time.monotonic() + _FFMPEG_BUDGET_SEC
 
     def _run_ffmpeg(codec: str) -> bool:
+        remaining = encode_deadline - time.monotonic()
+        if remaining <= 0:
+            print(f"[Storage] encode budget exhausted before {codec}")
+            return False
         cmd = ["ffmpeg", "-y", "-i", str(input_path)]
         if source_audio_path:
             cmd += ["-i", str(source_audio_path)]
@@ -46,10 +68,14 @@ def _run_ffmpeg(codec: str) -> bool:
                 check=True,
                 stdout=subprocess.DEVNULL,
                 stderr=subprocess.DEVNULL,
+                timeout=remaining,
             )
             return True
         except subprocess.CalledProcessError:
             return False
+        except subprocess.TimeoutExpired:
+            print(f"[Storage] ffmpeg ({codec}) hit the {_FFMPEG_BUDGET_SEC}s encode budget")
+            return False
 
     try:
         if _run_ffmpeg("h264_nvenc"):
@@ -74,25 +100,135 @@ def get_supabase():
     return create_client(url, key)
 
 
+def _b64(value: str) -> str:
+    return base64.b64encode(value.encode()).decode()
+
+
+def _permanent_status(exc: Exception) -> bool:
+    """True for a non-retryable 4xx, across httpx and the storage SDK.
+
+    The video (TUS) path raises ``httpx.HTTPStatusError``; the heatmap path raises
+    storage3's ``StorageApiError`` (status on ``.status``). 408/429 stay retryable.
+    """
+    if isinstance(exc, httpx.HTTPStatusError):
+        status = exc.response.status_code
+    else:
+        status = getattr(exc, "status", None)
+    try:
+        status = int(status)
+    except (TypeError, ValueError):
+        return False
+    return 400 <= status < 500 and status not in (408, 429)
+
+
+def _upload_with_retry(upload_fn, label: str, deadline: float):
+    """Run an upload with bounded exponential backoff and an overall time budget.
+
+    Supabase storage occasionally returns transient 5xx/504 under load; the video
+    upload had no retry, so a single blip failed the whole run. Retries stop once
+    ``deadline`` (monotonic) is reached so a degraded endpoint can't run out the
+    Modal job wall-clock. Permanent 4xx (e.g. 413) are not retried.
+    """
+    last_exc = None
+    for attempt in range(1, _UPLOAD_MAX_ATTEMPTS + 1):
+        try:
+            return upload_fn()
+        except Exception as exc:
+            if _permanent_status(exc):
+                raise
+            last_exc = exc
+            backoff = min(30, 2 ** attempt)
+            if attempt == _UPLOAD_MAX_ATTEMPTS or time.monotonic() + backoff >= deadline:
+                break
+            print(
+                f"[Storage] {label} upload attempt {attempt}/{_UPLOAD_MAX_ATTEMPTS} "
+                f"failed ({type(exc).__name__}: {exc}); retrying in {backoff}s"
+            )
+            time.sleep(backoff)
+    raise last_exc
+
+
+def _upload_resumable(
+    local_path: str,
+    bucket: str,
+    remote_path: str,
+    content_type: str,
+    deadline: float,
+    cache_control: str = "3600",
+) -> None:
+    """Upload a file via Supabase's resumable (TUS) endpoint in 6MB chunks.
+
+    Each chunk is its own request, so no single request approaches the standard
+    endpoint's ~50MB cap or storage3's 20s timeout. Every request's timeout is
+    capped by ``deadline`` (monotonic) so the whole upload stays within budget.
+    """
+    url = os.environ["SUPABASE_URL"].rstrip("/")
+    key = os.environ["SUPABASE_SERVICE_ROLE_KEY"]
+    size = os.path.getsize(local_path)
+    auth = {"Authorization": f"Bearer {key}", "apikey": key}
+    metadata = ",".join(
+        [
+            f"bucketName {_b64(bucket)}",
+            f"objectName {_b64(remote_path)}",
+            f"contentType {_b64(content_type)}",
+            f"cacheControl {_b64(cache_control)}",
+        ]
+    )
+
+    def _timeout() -> float:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise TimeoutError(f"upload budget exhausted for {remote_path}")
+        return min(_STORAGE_TIMEOUT_SEC, remaining)
+
+    with httpx.Client() as client:
+        created = client.post(
+            f"{url}/storage/v1/upload/resumable",
+            headers={
+                **auth,
+                "Tus-Resumable": "1.0.0",
+                "Upload-Length": str(size),
+                "Upload-Metadata": metadata,
+                "x-upsert": "true",
+            },
+            timeout=_timeout(),
+        )
+        created.raise_for_status()
+        location = created.headers["Location"]
+        if location.startswith("/"):
+            location = f"{url}{location}"
+
+        offset = 0
+        with open(local_path, "rb") as f:
+            while offset < size:
+                chunk = f.read(_TUS_CHUNK_SIZE)
+                patched = client.patch(
+                    location,
+                    headers={
+                        **auth,
+                        "Tus-Resumable": "1.0.0",
+                        "Upload-Offset": str(offset),
+                        "Content-Type": "application/offset+octet-stream",
+                    },
+                    content=chunk,
+                    timeout=_timeout(),
+                )
+                patched.raise_for_status()
+                offset = int(patched.headers["Upload-Offset"])
+
+
 def upload_processed_video(
     local_path: str,
     match_id: str,
     bucket: str = "results"
 ) -> str:
-    supabase = get_supabase()
     remote_path = f"{match_id}/processed.mp4"
-
-    with open(local_path, "rb") as f:
-        supabase.storage.from_(bucket).upload(
-            remote_path,
-            f,
-            file_options={
-                "content-type": "video/mp4",
-                "cacheControl": "3600",
-                "x-upsert": "true"
-            }
-        )
-
+    deadline = time.monotonic() + _UPLOAD_BUDGET_SEC
+    _upload_with_retry(
+        lambda: _upload_resumable(local_path, bucket, remote_path, "video/mp4", deadline),
+        label=f"processed video {match_id}",
+        deadline=deadline,
+    )
     return remote_path
 
 
@@ -103,20 +239,22 @@ def upload_heatmap_png(
     bucket: str = "results"
 ) -> str:
     """Upload a heatmap PNG to Supabase storage."""
-    supabase = get_supabase()
     remote_path = f"{match_id}/{filename}"
 
-    with open(local_path, "rb") as f:
-        supabase.storage.from_(bucket).upload(
-            remote_path,
-            f,
-            file_options={
-                "content-type": "image/png",
-                "cacheControl": "3600",
-                "x-upsert": "true"
-            }
-        )
+    def _do() -> None:
+        supabase = get_supabase()
+        with open(local_path, "rb") as f:
+            supabase.storage.from_(bucket).upload(
+                remote_path,
+                f,
+                file_options={
+                    "content-type": "image/png",
+                    "cacheControl": "3600",
+                    "x-upsert": "true",
+                },
+            )
 
+    _upload_with_retry(_do, label=f"{filename} {match_id}", deadline=time.monotonic() + _UPLOAD_BUDGET_SEC)
     return remote_path
 
 

backend/tests/test_storage_upload.py

@@ -5,6 +5,7 @@
 a match be marked "done" with no playable video. Heatmap uploads stay optional.
 """
 import pytest
+import httpx
 
 from backend.pipeline import storage
 
@@ -74,3 +75,64 @@ def test_all_uploads_succeed(monkeypatch, tmp_path):
     assert result["results_path"] == "match-123/processed.mp4"
     assert result["bounce_heatmap_path"] == "match-123/bounce_heatmap.png"
     assert result["player_heatmap_path"] == "match-123/player_heatmap.png"
+
+
+def test_upload_retries_transient_then_succeeds(monkeypatch):
+    """A transient error (timeout/connection blip) is retried, not fatal."""
+    monkeypatch.setattr(storage.time, "sleep", lambda *_: None)
+    calls = {"n": 0}
+
+    def _flaky():
+        calls["n"] += 1
+        if calls["n"] < 3:
+            raise httpx.ConnectError("transient blip")
+        return "done"
+
+    assert storage._upload_with_retry(_flaky, "test", storage.time.monotonic() + 60) == "done"
+    assert calls["n"] == 3
+
+
+def test_upload_does_not_retry_permanent_4xx(monkeypatch):
+    """A 413 (file over the project size limit) is permanent — fail on first try."""
+    monkeypatch.setattr(storage.time, "sleep", lambda *_: None)
+    calls = {"n": 0}
+
+    def _too_big():
+        calls["n"] += 1
+        request = httpx.Request("POST", "http://example/storage")
+        response = httpx.Response(413, request=request)
+        raise httpx.HTTPStatusError("payload too large", request=request, response=response)
+
+    with pytest.raises(httpx.HTTPStatusError):
+        storage._upload_with_retry(_too_big, "test", storage.time.monotonic() + 60)
+    assert calls["n"] == 1
+
+
+def test_upload_does_not_retry_permanent_storage_api_error(monkeypatch):
+    """A storage-SDK 4xx (e.g. 413 on a heatmap) is permanent — not retried."""
+    monkeypatch.setattr(storage.time, "sleep", lambda *_: None)
+    from storage3.exceptions import StorageApiError
+    calls = {"n": 0}
+
+    def _too_big():
+        calls["n"] += 1
+        raise StorageApiError("payload too large", "Payload too large", 413)
+
+    with pytest.raises(StorageApiError):
+        storage._upload_with_retry(_too_big, "test", storage.time.monotonic() + 60)
+    assert calls["n"] == 1
+
+
+def test_upload_stops_at_time_budget(monkeypatch):
+    """Retries stop once the overall deadline has passed, even for transient errors."""
+    monkeypatch.setattr(storage.time, "sleep", lambda *_: None)
+    calls = {"n": 0}
+
+    def _always_fails():
+        calls["n"] += 1
+        raise httpx.ConnectError("transient blip")
+
+    # Deadline already in the past -> one attempt, then stop (not all four).
+    with pytest.raises(httpx.ConnectError):
+        storage._upload_with_retry(_always_fails, "test", storage.time.monotonic() - 1)
+    assert calls["n"] == 1