fix(worker): add recovery and bounded retries

check_queue.py

@@ -11,14 +11,17 @@ c.table('saved_reels').delete().eq('url', 'https://www.instagram.com/reel/DIAGTE
 print('Test row deleted.')
 
 # Show any non-done rows
-res = c.table('saved_reels').select('id,url,status,source,created_at,error').in_('status', ['pending','processing','failed']).order('created_at', desc=True).limit(10).execute()
+res = c.table('saved_reels').select('id,url,status,source,created_at,error,attempt_count,processing_started_at,next_attempt_at').in_('status', ['pending','processing','failed','cookies_expired','unsupported_format']).order('created_at', desc=True).limit(10).execute()
 rows = res.data or []
 print(f'Non-done rows in queue: {len(rows)}')
 for r in rows:
     url = (r['url'] or 'NO URL')[:60]
     err = (r.get('error') or '')[:60]
     src = (r.get('source') or '')[:8]
-    print(f"  {r['status']:<12} | {src:<8} | {url} | err={err}")
+    attempts = r.get("attempt_count") or 0
+    started = r.get("processing_started_at") or "-"
+    next_at = r.get("next_attempt_at") or "-"
+    print(f"  {r['status']:<18} | tries={attempts} | started={started} | next={next_at} | {src:<8} | {url} | err={err}")
 
 # Also show total row count
 total = c.table('saved_reels').select('id', count='exact').execute()

db.py

@@ -111,17 +111,27 @@ def claim_next_pending() -> Optional[dict]:
     return the raw row. Returns None if the queue is empty or another worker
     won the claim."""
     c = get_client()
+    now = datetime.now(timezone.utc).isoformat()
     res = (c.table(TABLE).select("*")
-           .eq("status", "pending").order("created_at").limit(1).execute())
+           .eq("status", "pending")
+           .or_(f"next_attempt_at.is.null,next_attempt_at.lte.{now}")
+           .order("created_at").limit(1).execute())
     rows = res.data or []
     if not rows:
         return None
     row = rows[0]
-    upd = (c.table(TABLE).update({"status": "processing"})
+    attempt_count = int(row.get("attempt_count") or 0) + 1
+    payload = {
+        "status": "processing",
+        "processing_started_at": now,
+        "next_attempt_at": None,
+        "attempt_count": attempt_count,
+    }
+    upd = (c.table(TABLE).update(payload)
            .eq("id", row["id"]).eq("status", "pending").execute())
     if not upd.data:
         return None  # lost the race
-    return row
+    return {**row, **payload}
 
 
 def update_reel_result(reel_id: str, title: str, raw_transcript, post_caption,
@@ -134,6 +144,8 @@ def update_reel_result(reel_id: str, title: str, raw_transcript, post_caption,
         "extracted_json": extracted_json,
         "status": status,
         "error": None,
+        "processing_started_at": None,
+        "next_attempt_at": None,
     }).eq("id", reel_id).execute()
 
 
@@ -142,6 +154,8 @@ def mark_failed_with_status(reel_id: str, error, status: str = "failed") -> None
     get_client().table(TABLE).update({
         "status": status,
         "error": str(error)[:500],
+        "processing_started_at": None,
+        "next_attempt_at": None,
     }).eq("id", reel_id).execute()
 
 
@@ -150,6 +164,27 @@ def mark_failed(reel_id: str, error) -> None:
     mark_failed_with_status(reel_id, error, "failed")
 
 
+def recover_stale_processing(stale_before: datetime) -> int:
+    """Return abandoned claims older than stale_before to the queue."""
+    res = (get_client().table(TABLE).update({
+        "status": "pending",
+        "error": "Recovered stale processing claim",
+        "processing_started_at": None,
+    }).eq("status", "processing")
+      .lt("processing_started_at", stale_before.astimezone(timezone.utc).isoformat())
+      .execute())
+    return len(res.data or [])
+
+
+def schedule_retry(reel_id: str, error: str, next_attempt_at: datetime) -> None:
+    """Release a claim for a future retry while preserving its attempt count."""
+    get_client().table(TABLE).update({
+        "status": "pending",
+        "error": str(error)[:500],
+        "processing_started_at": None,
+        "next_attempt_at": next_attempt_at.astimezone(timezone.utc).isoformat(),
+    }).eq("id", reel_id).execute()
+
 def cluster_counts() -> list:
     res = get_client().table(TABLE).select("cluster").execute()
     counts = Counter((r.get("cluster") or "Unclustered") for r in (res.data or []))

main.py

@@ -5,14 +5,16 @@ import uuid
 import threading
 import tempfile
 import logging
-from datetime import datetime, timezone
+import random
+from datetime import datetime, timedelta, timezone
 from contextlib import asynccontextmanager
 from typing import List, Optional
 from fastapi import FastAPI, HTTPException, UploadFile, File, Form, Query, BackgroundTasks, Depends, Header
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel, Field, field_validator
 import yt_dlp
-from groq import Groq
+from groq import Groq, APIConnectionError, APIStatusError
+import httpx
 from dotenv import load_dotenv
 import re
 import glob
@@ -51,6 +53,21 @@ if not GROQ_API_KEY:
 
 API_AUTH_TOKEN = os.getenv("API_AUTH_TOKEN")
 
+
+def _positive_int_env(name: str, default: int) -> int:
+    try:
+        value = int(os.getenv(name, str(default)))
+        if value <= 0:
+            raise ValueError
+        return value
+    except ValueError:
+        logger.warning("%s must be a positive integer; using %s.", name, default)
+        return default
+
+
+WORKER_STALE_MINUTES = _positive_int_env("WORKER_STALE_MINUTES", 30)
+MAX_WORKER_ATTEMPTS = 3
+
 def verify_api_key(x_api_key: Optional[str] = Header(None)):
     """Access gate: enforce token check if API_AUTH_TOKEN environment variable is set."""
     if API_AUTH_TOKEN:
@@ -70,7 +87,6 @@ def init_local_db():
         logger.info("Supabase configuration detected.")
 
 groq_client = Groq(api_key=GROQ_API_KEY) if GROQ_API_KEY else None
-
 # Initialize main database
 init_local_db()
 
@@ -83,6 +99,10 @@ async def lifespan(app: FastAPI):
         else:
             logger.info("ENABLE_WORKER=0 — queue worker disabled.")
     else:
+        recovered = db.recover_stale_processing(
+            datetime.now(timezone.utc) - timedelta(minutes=WORKER_STALE_MINUTES)
+        )
+        logger.info("Recovered %s stale queue item(s).", recovered)
         threading.Thread(target=_worker_loop, daemon=True).start()
     yield
 
@@ -727,6 +747,8 @@ def process_pending_reel(row: dict) -> None:
         logger.info(f"Worker processed reel {reel_id} ({url})")
     except HTTPException as e:
         logger.warning(f"Worker failed reel {reel_id}: {e.detail}")
+        if is_retryable_worker_error(e) and _schedule_worker_retry(row, e.detail):
+            return
         status = "failed"
         if e.status_code == 403:
             status = "cookies_expired"
@@ -735,9 +757,38 @@ def process_pending_reel(row: dict) -> None:
         db.mark_failed_with_status(reel_id, e.detail, status)
     except Exception as e:
         logger.error(f"Worker error on reel {reel_id}: {str(e)}")
+        if is_retryable_worker_error(e) and _schedule_worker_retry(row, str(e)):
+            return
         db.mark_failed_with_status(reel_id, str(e), "failed")
 
 
+def is_retryable_worker_error(error: Exception) -> bool:
+    """Classify only known transient API and network failures as retryable."""
+    if isinstance(error, HTTPException):
+        return error.status_code == 429 or 500 <= error.status_code <= 599
+    if isinstance(error, APIStatusError):
+        return error.status_code == 429 or 500 <= error.status_code <= 599
+    return isinstance(error, (APIConnectionError, httpx.TimeoutException, httpx.NetworkError))
+
+
+def retry_delay_seconds(attempt_count: int, jitter: float | None = None) -> float:
+    """Return bounded exponential retry delay: about 30s, then about 2m."""
+    base = 30 * (4 ** max(0, attempt_count - 1))
+    factor = random.uniform(0.9, 1.1) if jitter is None else jitter
+    return base * factor
+
+
+def _schedule_worker_retry(row: dict, error: str) -> bool:
+    attempt_count = int(row.get("attempt_count") or 0)
+    if attempt_count >= MAX_WORKER_ATTEMPTS:
+        return False
+    next_attempt = datetime.now(timezone.utc) + timedelta(seconds=retry_delay_seconds(attempt_count))
+    db.schedule_retry(row["id"], error, next_attempt)
+    logger.warning("Scheduled retry %s/%s for reel %s at %s.",
+                   attempt_count + 1, MAX_WORKER_ATTEMPTS, row["id"], next_attempt.isoformat())
+    return True
+
+
 def worker_tick() -> bool:
     """Process at most one pending reel. Returns True if one was claimed."""
     row = db.claim_next_pending()
@@ -747,7 +798,7 @@ def worker_tick() -> bool:
     return True
 
 
-def _worker_loop(poll_interval: float = 5.0, idle_interval: float = 20.0) -> None:
+def _worker_loop(idle_interval: float = 20.0) -> None:
     """Background loop: drain the queue, backing off when it's empty."""
     logger.info("Queue worker started.")
     while True:
@@ -755,8 +806,10 @@ def _worker_loop(poll_interval: float = 5.0, idle_interval: float = 20.0) -> Non
             did_work = worker_tick()
         except Exception as e:
             logger.error(f"Worker tick crashed: {str(e)}")
-            did_work = False
-        time.sleep(poll_interval if did_work else idle_interval)
+            time.sleep(idle_interval)
+            continue
+        if not did_work:
+            time.sleep(idle_interval)
 
 
 @app.post("/extract/file", response_model=ExtractionResponse, dependencies=[Depends(verify_api_key)])
@@ -834,6 +887,10 @@ async def extract_text(payload: dict):
     return db_record
 
 
+ACTIVE_REEL_STATUSES = frozenset({"pending", "processing"})
+SUCCESS_REEL_STATUSES = frozenset({"done"})
+FAILURE_REEL_STATUSES = frozenset({"failed", "cookies_expired", "unsupported_format"})
+
 BATCH_JOB = {
     "status": "idle",   # idle | running | done | error
     "total": 0,
@@ -881,15 +938,18 @@ async def update_batch_job_status():
             processing += 1
         else:
             status = row.get("status") or "done"
-            if status == "done":
+            if status in SUCCESS_REEL_STATUSES:
                 ok += 1
-            elif status == "failed":
+            elif status in FAILURE_REEL_STATUSES:
                 failed += 1
                 errors.append({"url": url, "detail": row.get("error") or "Unknown error"})
-            else:
+            elif status in ACTIVE_REEL_STATUSES:
                 processing += 1
                 if status == "processing":
                     current_url = url
+            else:
+                failed += 1
+                errors.append({"url": url, "detail": f"Unknown queue status: {status}"})
                 
     done = ok + failed
     BATCH_JOB.update({
@@ -1002,4 +1062,3 @@ async def get_reels_status(payload: dict):
         db_rows.extend(res.data or [])
         
     return db_rows
-

run_worker.py

@@ -14,29 +14,26 @@ parent_env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(_
 if os.path.exists(parent_env_path):
     load_dotenv(parent_env_path)
 
-from main import worker_tick
+from datetime import datetime, timedelta, timezone
+import yt_dlp
+
+from main import WORKER_STALE_MINUTES, worker_tick
+from db import recover_stale_processing
 
 if __name__ == "__main__":
     logger.info("Standalone reels queue worker started locally.")
     
-    # Auto-update yt-dlp quietly at startup to bypass Meta/Instagram scraper blocks
-    logger.info("Checking for yt-dlp updates...")
-    try:
-        import subprocess
-        subprocess.run(
-            [sys.executable, "-m", "pip", "install", "--upgrade", "yt-dlp"],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            check=True
-        )
-        logger.info("yt-dlp is up to date.")
-    except Exception as ue:
-        logger.warning(f"Could not auto-update yt-dlp: {str(ue)}")
-        
+    logger.info("yt-dlp version: %s", getattr(yt_dlp.version, "__version__", "unknown"))
+
     if not os.getenv("SUPABASE_URL") or not os.getenv("SUPABASE_SERVICE_KEY"):
         logger.error("SUPABASE_URL / SUPABASE_SERVICE_KEY env vars not set. Exiting.")
         sys.exit(1)
         
+    recovered = recover_stale_processing(
+        datetime.now(timezone.utc) - timedelta(minutes=WORKER_STALE_MINUTES)
+    )
+    logger.info("Recovered %s stale queue item(s).", recovered)
+
     drain_mode = "--drain" in sys.argv
     if drain_mode:
         logger.info("Running in DRAIN mode. Will exit once queue is empty.")
@@ -54,5 +51,5 @@ if __name__ == "__main__":
             logger.info("No more pending reels in queue. Exiting drain worker.")
             break
             
-        time.sleep(5.0 if did_work else 20.0)
-
+        if not did_work:
+            time.sleep(20.0)

verify_pipeline.py

@@ -297,7 +297,7 @@ def test_worker_tick_failure():
     orig = (main.db.claim_next_pending, main.download_and_extract_audio, main.db.mark_failed, main.db.mark_failed_with_status)
     main.db.claim_next_pending = lambda: {"id": "row2", "url": "https://www.instagram.com/reel/F/"}
     def _boom(url):
-        raise HTTPException(status_code=500, detail="Meta blocking request")
+        raise HTTPException(status_code=400, detail="Meta blocking request")
     main.download_and_extract_audio = _boom
     
     mock_mark = lambda reel_id, error, status="failed": captured.update({"id": reel_id, "error": str(error), "status": status})
@@ -703,6 +703,103 @@ def test_api_auth_token_check():
         main.API_AUTH_TOKEN = orig_token
 
 
+
+def test_batch_terminal_statuses():
+    print("Testing all terminal and unknown batch statuses...")
+    import asyncio, main
+    original_client = main.db.get_client
+    previous = dict(main.BATCH_JOB)
+
+    class Query:
+        def select(self, *args): return self
+        def in_(self, column, values): return self
+        def execute(self):
+            return type("Result", (), {"data": [{"url": "u", "status": self.status, "error": "boom"}]})()
+
+    try:
+        for status in ("cookies_expired", "unsupported_format", "mystery"):
+            query = Query()
+            query.status = status
+            main.db.get_client = lambda q=query: type("Client", (), {"table": lambda self, name: q})()
+            main.BATCH_JOB.update({"status": "running", "urls": ["u"], "finished_at": None})
+            asyncio.run(main.update_batch_job_status())
+            assert main.BATCH_JOB["status"] == "done"
+            assert main.BATCH_JOB["failed"] == 1
+            if status == "mystery":
+                assert "Unknown queue status: mystery" in main.BATCH_JOB["errors"][0]["detail"]
+    finally:
+        main.db.get_client = original_client
+        main.BATCH_JOB.clear()
+        main.BATCH_JOB.update(previous)
+    print("[OK] terminal batch statuses passed!")
+
+
+def test_worker_retry_policy():
+    print("Testing bounded worker retry policy...")
+    import main, httpx
+    from fastapi import HTTPException
+    assert main.is_retryable_worker_error(HTTPException(status_code=429))
+    assert main.is_retryable_worker_error(HTTPException(status_code=503))
+    assert main.is_retryable_worker_error(httpx.ConnectError("offline"))
+    assert not main.is_retryable_worker_error(HTTPException(status_code=400))
+    assert not main.is_retryable_worker_error(HTTPException(status_code=403))
+    assert not main.is_retryable_worker_error(HTTPException(status_code=415))
+    assert 27 <= main.retry_delay_seconds(1, .9) <= 33
+    assert 108 <= main.retry_delay_seconds(2, .9) <= 132
+    calls = []
+    original = main.db.schedule_retry
+    main.db.schedule_retry = lambda reel_id, error, when: calls.append((reel_id, error, when))
+    try:
+        assert main._schedule_worker_retry({"id": "a", "attempt_count": 1}, "temporary")
+        assert main._schedule_worker_retry({"id": "b", "attempt_count": 2}, "temporary")
+        assert not main._schedule_worker_retry({"id": "c", "attempt_count": 3}, "temporary")
+        assert [call[0] for call in calls] == ["a", "b"]
+    finally:
+        main.db.schedule_retry = original
+    print("[OK] bounded retry policy passed!")
+
+
+def test_db_queue_metadata_contract():
+    print("Testing queue metadata data-layer contract...")
+    import db
+    from datetime import datetime, timezone
+
+    class Result:
+        def __init__(self, data): self.data = data
+    class Query:
+        def __init__(self):
+            self.calls = []
+            self.selected_rows = [{"id": "r1", "attempt_count": 1}]
+            self.updated = None
+        def select(self, *args): self.calls.append(("select", args)); return self
+        def update(self, payload): self.updated = payload; self.calls.append(("update", payload)); return self
+        def eq(self, *args): self.calls.append(("eq", args)); return self
+        def or_(self, value): self.calls.append(("or", value)); return self
+        def order(self, *args, **kwargs): return self
+        def limit(self, value): return self
+        def lt(self, *args): self.calls.append(("lt", args)); return self
+        def execute(self):
+            return Result(self.selected_rows if self.updated is None else [{"id": "r1"}])
+
+    query = Query()
+    original = db.get_client
+    db.get_client = lambda: type("Client", (), {"table": lambda self, name: query})()
+    try:
+        claimed = db.claim_next_pending()
+        assert claimed["attempt_count"] == 2
+        assert any(call[0] == "or" and "next_attempt_at.is.null" in call[1] and "next_attempt_at.lte." in call[1] for call in query.calls)
+        assert query.updated["status"] == "processing"
+        assert query.updated["processing_started_at"]
+        assert query.updated["next_attempt_at"] is None
+        query.calls.clear()
+        assert db.recover_stale_processing(datetime.now(timezone.utc)) == 1
+        assert ("eq", ("status", "processing")) in query.calls
+        assert any(call[0] == "lt" and call[1][0] == "processing_started_at" for call in query.calls)
+    finally:
+        db.get_client = original
+    print("[OK] queue metadata contract passed!")
+
+
 if __name__ == "__main__":
     print("--- Starting Transcriber Pipeline Test ---")
     test_health()
@@ -734,8 +831,8 @@ if __name__ == "__main__":
     test_worker_tick_invalid_url()
     test_download_cleanup_on_failure()
     test_api_auth_token_check()
+    test_batch_terminal_statuses()
+    test_worker_retry_policy()
+    test_db_queue_metadata_contract()
     print("--- All tests completed successfully! ---")
     sys.exit(0)
-
-
-