From d46b32c053558400b8c33e81aa0791c536936624 Mon Sep 17 00:00:00 2001 From: Anik Majumdar Date: Sat, 12 Sep 2026 11:53:36 -0700 Subject: [PATCH 1/2] fix: upload processed video via resumable endpoint with retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The finalize-stage processed-video upload failed for larger clips (verified same footage: a 2s clip uploads fine, the 50s/~16MB version fails 2/2), marking the whole match failed with num_frames/fps left NULL. Root cause: the single-shot standard /object upload is bounded by storage3's 20s-per-request timeout — a ~16MB upload over a slow/contended Modal->Supabase egress exceeds it — and, unlike the download path, the upload had no retry. - Upload processed.mp4 via Supabase's resumable (TUS) endpoint in 6MB chunks, so no single request approaches the 20s timeout. - Add bounded exponential-backoff retry to the video and heatmap uploads; skip retry on permanent 4xx (e.g. 413). - Add an ffmpeg subprocess timeout in make_streamable_mp4 so a hung encode fails fast instead of running to the Modal wall-clock ("stopped during finalize"). Verified against Supabase: 16MB and 48MB upload via resumable; 60MB fails fast with a single 413 (files >~50MB additionally need the project's global upload size limit raised in Supabase settings). --- backend/pipeline/storage.py | 150 ++++++++++++++++++++++----- backend/tests/test_storage_upload.py | 32 ++++++ 2 files changed, 158 insertions(+), 24 deletions(-) diff --git a/backend/pipeline/storage.py b/backend/pipeline/storage.py index 93d15f6a..9558b908 100644 --- a/backend/pipeline/storage.py +++ b/backend/pipeline/storage.py @@ -1,9 +1,23 @@ 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 +_FFMPEG_TIMEOUT_SEC = 900 # cap a hung encode instead of running to the Modal wall-clock + 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). @@ -46,10 +60,14 @@ def _run_ffmpeg(codec: str) -> bool: check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + timeout=_FFMPEG_TIMEOUT_SEC, ) return True except subprocess.CalledProcessError: return False + except subprocess.TimeoutExpired: + print(f"[Storage] ffmpeg ({codec}) timed out after {_FFMPEG_TIMEOUT_SEC}s — trying fallback") + return False try: if _run_ffmpeg("h264_nvenc"): @@ -74,25 +92,107 @@ def get_supabase(): return create_client(url, key) +def _b64(value: str) -> str: + return base64.b64encode(value.encode()).decode() + + +def _upload_with_retry(upload_fn, label: str): + """Run an upload with bounded exponential backoff. + + Supabase storage occasionally returns transient 5xx/504 under load; the video + upload had no retry, so a single blip failed the whole run. Permanent 4xx + (e.g. 413 when a file exceeds the project's global size limit) are not retried. + """ + last_exc = None + for attempt in range(1, _UPLOAD_MAX_ATTEMPTS + 1): + try: + return upload_fn() + except Exception as exc: + if isinstance(exc, httpx.HTTPStatusError): + status = exc.response.status_code + if 400 <= status < 500 and status not in (408, 429): + raise + last_exc = exc + if attempt == _UPLOAD_MAX_ATTEMPTS: + break + backoff = min(30, 2 ** attempt) + 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, + 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. + """ + 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)}", + ] + ) + + with httpx.Client(timeout=httpx.Timeout(_STORAGE_TIMEOUT_SEC)) 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", + }, + ) + 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, + ) + 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" - } - ) - + _upload_with_retry( + lambda: _upload_resumable(local_path, bucket, remote_path, "video/mp4"), + label=f"processed video {match_id}", + ) return remote_path @@ -103,20 +203,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}") return remote_path diff --git a/backend/tests/test_storage_upload.py b/backend/tests/test_storage_upload.py index d5cb654e..da8a87b4 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,34 @@ 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") == "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") + assert calls["n"] == 1 From 55fffaab1dee00b11d3c116550a4c0e8d81ef6af Mon Sep 17 00:00:00 2001 From: Anik Majumdar Date: Sat, 12 Sep 2026 12:34:43 -0700 Subject: [PATCH 2/2] fix: bound encode/upload deadlines and classify SDK 4xx as permanent Addresses PR #86 review comments: - make_streamable_mp4 shares one encode deadline across both codec attempts, so a hung nvenc followed by a hung libx264 can no longer each burn the full budget (previously up to 2x900s = the whole 1800s job). - Upload retries are bounded by an overall 600s per-artifact budget: the deadline gates both the retry loop and each TUS chunk's request timeout, so a degraded endpoint cannot run out the 1800s process_video wall-clock. - _permanent_status classifies storage3 StorageApiError 4xx (e.g. a heatmap 413) as permanent, not just httpx.HTTPStatusError, so SDK-path 4xx aren't retried. Adds tests for SDK 4xx classification and the time-budget cutoff (7 pass). --- backend/pipeline/storage.py | 72 +++++++++++++++++++++------- backend/tests/test_storage_upload.py | 34 ++++++++++++- 2 files changed, 86 insertions(+), 20 deletions(-) diff --git a/backend/pipeline/storage.py b/backend/pipeline/storage.py index 9558b908..dca7e006 100644 --- a/backend/pipeline/storage.py +++ b/backend/pipeline/storage.py @@ -15,8 +15,9 @@ # 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 -_FFMPEG_TIMEOUT_SEC = 900 # cap a hung encode instead of running to the Modal wall-clock +_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: """ @@ -34,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)] @@ -60,13 +68,13 @@ def _run_ffmpeg(codec: str) -> bool: check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, - timeout=_FFMPEG_TIMEOUT_SEC, + timeout=remaining, ) return True except subprocess.CalledProcessError: return False except subprocess.TimeoutExpired: - print(f"[Storage] ffmpeg ({codec}) timed out after {_FFMPEG_TIMEOUT_SEC}s — trying fallback") + print(f"[Storage] ffmpeg ({codec}) hit the {_FFMPEG_BUDGET_SEC}s encode budget") return False try: @@ -96,26 +104,42 @@ def _b64(value: str) -> str: return base64.b64encode(value.encode()).decode() -def _upload_with_retry(upload_fn, label: str): - """Run an upload with bounded exponential backoff. +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. Permanent 4xx - (e.g. 413 when a file exceeds the project's global size limit) are not retried. + 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 isinstance(exc, httpx.HTTPStatusError): - status = exc.response.status_code - if 400 <= status < 500 and status not in (408, 429): - raise + if _permanent_status(exc): + raise last_exc = exc - if attempt == _UPLOAD_MAX_ATTEMPTS: - break 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" @@ -129,12 +153,14 @@ def _upload_resumable( 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. + 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"] @@ -149,7 +175,13 @@ def _upload_resumable( ] ) - with httpx.Client(timeout=httpx.Timeout(_STORAGE_TIMEOUT_SEC)) as client: + 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={ @@ -159,6 +191,7 @@ def _upload_resumable( "Upload-Metadata": metadata, "x-upsert": "true", }, + timeout=_timeout(), ) created.raise_for_status() location = created.headers["Location"] @@ -178,6 +211,7 @@ def _upload_resumable( "Content-Type": "application/offset+octet-stream", }, content=chunk, + timeout=_timeout(), ) patched.raise_for_status() offset = int(patched.headers["Upload-Offset"]) @@ -189,9 +223,11 @@ def upload_processed_video( bucket: str = "results" ) -> str: remote_path = f"{match_id}/processed.mp4" + deadline = time.monotonic() + _UPLOAD_BUDGET_SEC _upload_with_retry( - lambda: _upload_resumable(local_path, bucket, remote_path, "video/mp4"), + lambda: _upload_resumable(local_path, bucket, remote_path, "video/mp4", deadline), label=f"processed video {match_id}", + deadline=deadline, ) return remote_path @@ -218,7 +254,7 @@ def _do() -> None: }, ) - _upload_with_retry(_do, label=f"{filename} {match_id}") + _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 da8a87b4..1d4211aa 100644 --- a/backend/tests/test_storage_upload.py +++ b/backend/tests/test_storage_upload.py @@ -88,7 +88,7 @@ def _flaky(): raise httpx.ConnectError("transient blip") return "done" - assert storage._upload_with_retry(_flaky, "test") == "done" + assert storage._upload_with_retry(_flaky, "test", storage.time.monotonic() + 60) == "done" assert calls["n"] == 3 @@ -104,5 +104,35 @@ def _too_big(): raise httpx.HTTPStatusError("payload too large", request=request, response=response) with pytest.raises(httpx.HTTPStatusError): - storage._upload_with_retry(_too_big, "test") + 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