diff --git a/backend/pipeline/storage.py b/backend/pipeline/storage.py index 93d15f6a..dca7e006 100644 --- a/backend/pipeline/storage.py +++ b/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 diff --git a/backend/tests/test_storage_upload.py b/backend/tests/test_storage_upload.py index d5cb654e..1d4211aa 100644 --- a/backend/tests/test_storage_upload.py +++ b/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