fix: harden publish retry and proxy checks

pyproject.toml

@@ -18,7 +18,7 @@ dependencies = [
   "sqlalchemy>=2.0.30",
   "psycopg[binary]>=3.2.0",
   "boto3>=1.34.0",
-  "httpx>=0.27.0",
+  "httpx[socks]>=0.27.0",
   "python-multipart>=0.0.9",
   "celery>=5.4.0",
   "redis>=5.0.0",

src/videoroll/apps/bilibili_publisher/main.py

@@ -41,6 +41,10 @@ from videoroll.apps.subtitle_service.bilibili_tags_store import get_task_bilibil
 from videoroll.apps.subtitle_service.translate_settings_store import get_translate_settings
 
 
+def _as_dict(v: object) -> dict:
+    return v if isinstance(v, dict) else {}
+
+
 def get_settings() -> BilibiliPublisherSettings:
     return get_bilibili_publisher_settings()
 
@@ -223,12 +227,47 @@ def _utcnow() -> datetime:
     return datetime.now(tz=timezone.utc)
 
 
+def _latest_publish_job(db: Session, task_id: uuid.UUID, states: set[PublishState]) -> PublishJob | None:
+    return (
+        db.query(PublishJob)
+        .filter(PublishJob.task_id == task_id, PublishJob.state.in_(list(states)))
+        .order_by(PublishJob.updated_at.desc(), PublishJob.created_at.desc())
+        .first()
+    )
+
+
+def _publish_response_from_job(job: PublishJob) -> PublishResponse:
+    return PublishResponse(
+        state=job.state.value,
+        aid=job.aid,
+        bvid=job.bvid,
+        response=_as_dict(job.response_json) or None,
+    )
+
+
 @app.post("/bilibili/publish", response_model=PublishResponse)
 def publish(payload: PublishRequest, settings: BilibiliPublisherSettings = Depends(get_settings), db: Session = Depends(get_db)) -> PublishResponse:
-    task = db.get(Task, payload.task_id)
+    task = db.get(Task, payload.task_id, with_for_update=True)
     if not task:
         raise HTTPException(status_code=404, detail="task not found")
 
+    published_job = _latest_publish_job(db, payload.task_id, {PublishState.published})
+    if published_job:
+        task.status = TaskStatus.published
+        task.error_code = None
+        task.error_message = None
+        db.add(task)
+        db.commit()
+        return _publish_response_from_job(published_job)
+
+    active_job = _latest_publish_job(db, payload.task_id, {PublishState.submitting, PublishState.submitted})
+    if active_job:
+        if settings.publish_mode != "mock":
+            task.status = TaskStatus.publishing
+            db.add(task)
+            db.commit()
+        return _publish_response_from_job(active_job)
+
     job = PublishJob(
         task_id=payload.task_id,
         meta_json={

src/videoroll/apps/bilibili_publisher/worker.py

@@ -350,6 +350,28 @@ def _publish_stage_lock_key(stage: str) -> str:
     return f"videoroll:bilibili:publish:{stage}:lock"
 
 
+def _publish_job_lock_key(job_id: str) -> str:
+    return f"videoroll:bilibili:publish:job:{job_id}:lock"
+
+
+def _try_acquire_publish_job_lock(job_id: str) -> Any:
+    client = _redis_client()
+    if client is None:
+        return None
+    lock = client.lock(
+        _publish_job_lock_key(job_id),
+        timeout=int(_env_float("BILIBILI_PUBLISH_JOB_LOCK_TIMEOUT_SECONDS", 3600.0, minimum=60.0)),
+        blocking_timeout=0,
+    )
+    try:
+        if not bool(lock.acquire(blocking=False)):
+            return False
+    except RedisError as e:
+        logger.warning("publish job lock disabled for this attempt (job_id=%s): %s", job_id, e)
+        return None
+    return lock
+
+
 def _parse_float(raw: Any) -> float | None:
     if raw is None:
         return None
@@ -501,6 +523,38 @@ def _publish_meta_with_retry_desc_limit(meta: BilibiliPublishMeta) -> BilibiliPu
     return BilibiliPublishMeta.model_validate(payload)
 
 
+def _latest_published_job(db: Session, task_id: uuid.UUID) -> PublishJob | None:
+    return (
+        db.query(PublishJob)
+        .filter(PublishJob.task_id == task_id, PublishJob.state == PublishState.published)
+        .order_by(PublishJob.updated_at.desc(), PublishJob.created_at.desc())
+        .first()
+    )
+
+
+def _task_has_published_job(db: Session, task_id: uuid.UUID) -> bool:
+    return _latest_published_job(db, task_id) is not None
+
+
+def _can_mark_task_publish_failed(db: Session, task: Task) -> bool:
+    return task.status != TaskStatus.published and not _task_has_published_job(db, task.id)
+
+
+def _mirror_published_job(job: PublishJob, published_job: PublishJob | None, task: Task) -> None:
+    job.state = PublishState.published
+    job.aid = published_job.aid if published_job else job.aid
+    job.bvid = published_job.bvid if published_job else job.bvid
+    job.response_json = {
+        **_as_dict(published_job.response_json if published_job else job.response_json),
+        "skipped_duplicate_publish": True,
+        "reason": "task already published",
+    }
+    job.updated_at = _utcnow()
+    task.status = TaskStatus.published
+    task.error_code = None
+    task.error_message = None
+
+
 def _add_archive_with_desc_retry(
     *,
     client: BilibiliWebClient,
@@ -541,8 +595,13 @@ def process_job(self, job_id: str) -> dict[str, Any]:
     store: S3Store | None = None
     upload_throttle: dict[str, Any] | None = None
     submit_throttle: dict[str, Any] | None = None
+    job_lock: Any = None
     try:
         job_uuid = uuid.UUID(job_id)
+        job_lock = _try_acquire_publish_job_lock(job_id)
+        if job_lock is False:
+            return {"status": "skipped", "detail": "publish job already running"}
+
         job = db.get(PublishJob, job_uuid)
         if not job:
             return {"status": "not_found"}
@@ -559,6 +618,14 @@ def process_job(self, job_id: str) -> dict[str, Any]:
             db.commit()
             return {"status": "error", "detail": "task not found"}
 
+        published_job = _latest_published_job(db, task.id)
+        if task.status == TaskStatus.published or published_job is not None:
+            _mirror_published_job(job, published_job, task)
+            db.add(job)
+            db.add(task)
+            db.commit()
+            return {"status": "skipped", "detail": "task already published", "aid": job.aid, "bvid": job.bvid}
+
         cookie = (get_bilibili_cookie_header(db) or "").strip()
         csrf = (get_bilibili_csrf_token(db) or "").strip()
         if not cookie:
@@ -886,7 +953,7 @@ def process_job(self, job_id: str) -> dict[str, Any]:
                     db.add(job)
                 if task is None and job:
                     task = db.get(Task, job.task_id)
-                if task:
+                if task and _can_mark_task_publish_failed(db, task):
                     task.status = TaskStatus.failed
                     task.error_code = "PUBLISH_RATE_LIMITED"
                     task.error_message = e.message or str(e)
@@ -923,7 +990,7 @@ def process_job(self, job_id: str) -> dict[str, Any]:
                 db.add(job)
             if task is None and job:
                 task = db.get(Task, job.task_id)
-            if task:
+            if task and _can_mark_task_publish_failed(db, task):
                 task.status = TaskStatus.failed
                 task.error_code = "PUBLISH_FAILED"
                 task.error_message = str(e)
@@ -945,4 +1012,9 @@ def process_job(self, job_id: str) -> dict[str, Any]:
             pass
         return {"status": "error", "detail": str(e)}
     finally:
+        if job_lock not in (None, False):
+            try:
+                job_lock.release()
+            except Exception:
+                pass
         db.close()

src/videoroll/apps/orchestrator_api/main.py

@@ -36,9 +36,11 @@ from videoroll.db.models import (
     AssetKind,
     IngestedVideo,
     PublishJob,
+    PublishState,
     RenderJob,
     RenderJobStatus,
     SourceLicense,
+    SourceType,
     Subtitle,
     SubtitleJob,
     SubtitleJobStatus,
@@ -49,6 +51,7 @@ from videoroll.db.session import db_session, get_engine, get_sessionmaker
 from videoroll.storage.s3 import S3Store
 from videoroll.utils.auto_youtube import encode_auto_youtube_created_by
 from videoroll.utils.hashing import sha256_file
+from videoroll.utils.httpx_proxy import HTTPX_PROXY_KWARG_UNSUPPORTED, format_httpx_proxy_error
 from videoroll.utils.internal_api_token import internal_api_token
 from videoroll.utils.workdir_maintenance import (
     WorkdirJobState,
@@ -206,6 +209,34 @@ def _publish_job_error_message(job: PublishJob) -> str | None:
     return msg
 
 
+def _published_publish_job_task_ids(db: Session, task_ids: list[uuid.UUID]) -> set[uuid.UUID]:
+    if not task_ids:
+        return set()
+    rows = (
+        db.query(PublishJob.task_id)
+        .filter(PublishJob.task_id.in_(task_ids), PublishJob.state == PublishState.published)
+        .all()
+    )
+    return {row[0] for row in rows}
+
+
+def _reconcile_published_task_state(db: Session, task: Task, *, published_task_ids: set[uuid.UUID] | None = None) -> bool:
+    if task.status in {TaskStatus.published, TaskStatus.canceled}:
+        return False
+    has_published_job = task.id in published_task_ids if published_task_ids is not None else bool(
+        db.query(PublishJob.id)
+        .filter(PublishJob.task_id == task.id, PublishJob.state == PublishState.published)
+        .first()
+    )
+    if not has_published_job:
+        return False
+    task.status = TaskStatus.published
+    task.error_code = None
+    task.error_message = None
+    db.add(task)
+    return True
+
+
 def _normalize_tags(v: Any) -> list[str]:
     if v is None:
         return []
@@ -1945,10 +1976,19 @@ def list_tasks(
     tasks = q.limit(limit).all()
 
     ids = [t.id for t in tasks]
+    published_task_ids = _published_publish_job_task_ids(db, ids)
+    reconciled = False
+    for t in tasks:
+        reconciled = _reconcile_published_task_state(db, t, published_task_ids=published_task_ids) or reconciled
+    if reconciled:
+        db.commit()
+
     title_map = _load_task_display_titles(db, ids, s3=s3, allow_s3_fallback=True)
 
     out: list[dict[str, Any]] = []
     for t in tasks:
+        if status is not None and t.status != status:
+            continue
         item = TaskRead.model_validate(t).model_dump()
         display_title = str(title_map.get(t.id) or "").strip()
         item["display_title"] = display_title or None
@@ -2121,7 +2161,7 @@ def test_youtube_proxy(
                     used_proxy=None,
                     status_code=resp.status_code,
                     elapsed_ms=elapsed_ms,
-                    error="httpx does not support proxy kwarg in this environment",
+                    error=HTTPX_PROXY_KWARG_UNSUPPORTED,
                 )
     except Exception as e:
         elapsed_ms = int((time.perf_counter() - start) * 1000)
@@ -2131,7 +2171,7 @@ def test_youtube_proxy(
             used_proxy=proxy or None,
             status_code=None,
             elapsed_ms=elapsed_ms,
-            error=str(e),
+            error=format_httpx_proxy_error(e, proxy=proxy),
         )
 
 
@@ -2140,6 +2180,8 @@ def get_task(task_id: uuid.UUID, db: Session = Depends(get_db), s3: S3Store = De
     task = db.get(Task, task_id)
     if not task:
         raise HTTPException(status_code=404, detail="task not found")
+    if _reconcile_published_task_state(db, task):
+        db.commit()
     item = TaskRead.model_validate(task).model_dump()
     title = get_task_display_title_with_s3(db, str(task_id), s3=s3).strip()
     item["display_title"] = title or None

src/videoroll/apps/subtitle_service/main.py

@@ -60,6 +60,7 @@ from videoroll.apps.subtitle_service.worker_concurrency import sync_subtitle_wor
 from videoroll.apps.subtitle_service.worker import TASK_QUEUE_LOCK_OWNER, celery_app
 from videoroll.utils.auto_youtube import parse_auto_youtube_created_by
 from videoroll.utils.cpu import process_cpu_count
+from videoroll.utils.httpx_proxy import HTTPX_PROXY_KWARG_UNSUPPORTED, format_httpx_proxy_error
 from videoroll.utils.intel_gpu import detect_intel_hardware
 
 logger = logging.getLogger(__name__)
@@ -449,7 +450,7 @@ def test_model_download_proxy(
                     used_proxy=None,
                     status_code=resp.status_code,
                     elapsed_ms=elapsed_ms,
-                    error="httpx does not support proxy kwarg in this environment",
+                    error=HTTPX_PROXY_KWARG_UNSUPPORTED,
                 )
     except Exception as e:
         elapsed_ms = int((time.perf_counter() - start) * 1000)
@@ -459,7 +460,7 @@ def test_model_download_proxy(
             used_proxy=proxy or None,
             status_code=None,
             elapsed_ms=elapsed_ms,
-            error=str(e),
+            error=format_httpx_proxy_error(e, proxy=proxy),
         )
 
 

src/videoroll/db/auto_migrate.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+import os
 from functools import lru_cache
 
 from sqlalchemy import inspect, text
@@ -102,9 +103,13 @@ def auto_migrate_engine(engine: Engine) -> None:
 
 
 @lru_cache
-def auto_migrate(database_url: str) -> None:
+def _auto_migrate_cached(database_url: str, pid: int) -> None:
     """
-    Idempotent auto migration (cached per database_url).
+    Idempotent auto migration cached per process.
     """
     engine = get_engine(database_url)
     auto_migrate_engine(engine)
+
+
+def auto_migrate(database_url: str) -> None:
+    _auto_migrate_cached(database_url, os.getpid())

src/videoroll/db/session.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 from functools import lru_cache
 from typing import Generator
 
@@ -21,10 +22,16 @@ def _engine_connect_args(database_url: str) -> dict[str, object]:
 
 
 @lru_cache
-def get_engine(database_url: str) -> Engine:
+def _get_engine_cached(database_url: str, pid: int) -> Engine:
     return create_engine(database_url, pool_pre_ping=True, connect_args=_engine_connect_args(database_url))
 
 
+def get_engine(database_url: str) -> Engine:
+    # Celery prefork workers inherit module globals from the parent process.
+    # Keep engines isolated per PID so child processes don't reuse the parent's pool.
+    return _get_engine_cached(database_url, os.getpid())
+
+
 def get_sessionmaker(database_url: str) -> sessionmaker[Session]:
     engine = get_engine(database_url)
     return sessionmaker(bind=engine, autocommit=False, autoflush=False)

src/videoroll/utils/httpx_proxy.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from urllib.parse import urlsplit
+
+HTTPX_PROXY_KWARG_UNSUPPORTED = "The current backend httpx build does not support the `proxy` argument."
+HTTPX_SOCKS_SUPPORT_MISSING = (
+    "SOCKS proxy support is unavailable in the current backend environment. "
+    "Install `httpx[socks]` (includes `socksio`) or use an `http://` proxy."
+)
+
+
+def uses_socks_proxy(proxy: str | None) -> bool:
+    scheme = urlsplit(str(proxy or "").strip()).scheme.lower()
+    return scheme.startswith("socks")
+
+
+def format_httpx_proxy_error(error: Exception, *, proxy: str | None) -> str:
+    message = str(error).strip() or type(error).__name__
+    lower_message = message.lower()
+    if uses_socks_proxy(proxy) and (
+        "socksio" in lower_message
+        or "httpx[socks]" in lower_message
+        or "using socks proxy" in lower_message
+    ):
+        return HTTPX_SOCKS_SUPPORT_MISSING
+    return message

src/web/src/pages/SettingsYouTubePage.tsx

@@ -352,14 +352,19 @@ export default function SettingsYouTubePage() {
       <div className="rounded border bg-white p-4">
         <div className="text-sm font-semibold">代理设置</div>
         <div className="mt-2 text-xs text-slate-500">
-          例如：<span className="font-mono">http://127.0.0.1:7890</span> / <span className="font-mono">socks5://127.0.0.1:1080</span>（留空=不使用代理）
+          支持 <span className="font-mono">http://</span> 和 <span className="font-mono">socks5://</span>，例如
+          <span className="font-mono"> http://127.0.0.1:7890 </span> /
+          <span className="font-mono"> socks5://127.0.0.1:1080</span>（留空 = 不使用代理）
+        </div>
+        <div className="mt-1 text-xs text-slate-500">
+          SOCKS5 检测和抓取依赖后端安装 SOCKS 支持；当前生产镜像已内置该依赖。
         </div>
         <div className="mt-3 grid gap-3 md:grid-cols-2">
           <label className="block md:col-span-2">
             <div className="mb-1 text-xs text-slate-600">proxy</div>
             <input
               className="w-full rounded border px-3 py-2 text-sm"
-              placeholder="http://host:port"
+              placeholder="http://host:port 或 socks5://host:port"
               value={proxy}
               onChange={(e) => setProxy(e.target.value)}
             />

tests/test_bilibili_publish_idempotency.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import uuid
+from unittest.mock import patch
+
+from videoroll.apps.bilibili_publisher import worker
+from videoroll.db.models import PublishJob, PublishState, SourceLicense, SourceType, Task, TaskStatus
+
+
+def _task(*, status: TaskStatus = TaskStatus.publishing) -> Task:
+    return Task(
+        id=uuid.uuid4(),
+        source_type=SourceType.youtube,
+        source_license=SourceLicense.own,
+        status=status,
+        error_code="PUBLISH_FAILED",
+        error_message="old failure",
+    )
+
+
+def test_mirror_published_job_marks_duplicate_as_published_without_resubmitting() -> None:
+    task = _task(status=TaskStatus.failed)
+    published = PublishJob(
+        id=uuid.uuid4(),
+        task_id=task.id,
+        state=PublishState.published,
+        aid="123",
+        bvid="BV123",
+        response_json={"mode": "web", "result": {"aid": "123", "bvid": "BV123"}},
+    )
+    duplicate = PublishJob(id=uuid.uuid4(), task_id=task.id, state=PublishState.submitting)
+
+    worker._mirror_published_job(duplicate, published, task)
+
+    assert duplicate.state == PublishState.published
+    assert duplicate.aid == "123"
+    assert duplicate.bvid == "BV123"
+    assert duplicate.response_json["skipped_duplicate_publish"] is True
+    assert task.status == TaskStatus.published
+    assert task.error_code is None
+    assert task.error_message is None
+
+
+def test_published_task_cannot_be_marked_publish_failed() -> None:
+    task = _task(status=TaskStatus.published)
+
+    assert worker._can_mark_task_publish_failed(None, task) is False  # type: ignore[arg-type]
+
+
+def test_task_with_published_job_cannot_be_marked_publish_failed() -> None:
+    task = _task(status=TaskStatus.publishing)
+
+    with patch.object(worker, "_task_has_published_job", return_value=True):
+        assert worker._can_mark_task_publish_failed(None, task) is False  # type: ignore[arg-type]

tests/test_bilibili_publish_rate_limit.py

@@ -9,19 +9,25 @@ from videoroll.apps.bilibili_publisher import worker
 
 
 class _FakeLock:
+    def __init__(self, acquired: bool = True) -> None:
+        self.acquired = acquired
+        self.released = False
+
     def acquire(self, blocking: bool = True) -> bool:
-        return True
+        return self.acquired
 
     def release(self) -> None:
+        self.released = True
         return None
 
 
 class _FakeRedis:
-    def __init__(self) -> None:
+    def __init__(self, *, lock_acquired: bool = True) -> None:
+        self.lock_acquired = lock_acquired
         self.store: dict[str, str] = {}
 
     def lock(self, *args, **kwargs) -> _FakeLock:
-        return _FakeLock()
+        return _FakeLock(acquired=self.lock_acquired)
 
     def get(self, key: str) -> str | None:
         return self.store.get(key)
@@ -74,3 +80,17 @@ def test_reserve_publish_stage_slot_uses_queue_depth_to_push_later_calls() -> No
     assert first["interval_seconds"] == 75.0
     assert second["wait_seconds"] == 75.0
     assert second["source"] == "redis"
+
+
+def test_try_acquire_publish_job_lock_returns_false_when_already_running() -> None:
+    fake_redis = _FakeRedis(lock_acquired=False)
+    with patch.object(worker, "_redis_client", return_value=fake_redis):
+        assert worker._try_acquire_publish_job_lock("job-1") is False
+
+
+def test_try_acquire_publish_job_lock_returns_lock_when_acquired() -> None:
+    fake_redis = _FakeRedis(lock_acquired=True)
+    with patch.object(worker, "_redis_client", return_value=fake_redis):
+        lock = worker._try_acquire_publish_job_lock("job-1")
+
+    assert isinstance(lock, _FakeLock)

tests/test_db_session_pid_isolation.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from videoroll.db import auto_migrate as auto_migrate_module
+from videoroll.db import session as session_module
+
+
+def test_get_engine_isolated_per_pid(monkeypatch) -> None:
+    session_module._get_engine_cached.cache_clear()
+    calls: list[tuple[str, bool, dict[str, object]]] = []
+
+    def fake_create_engine(url: str, *, pool_pre_ping: bool, connect_args: dict[str, object]):
+        engine = object()
+        calls.append((url, pool_pre_ping, connect_args))
+        return engine
+
+    monkeypatch.setattr(session_module, "create_engine", fake_create_engine)
+    monkeypatch.setattr(session_module.os, "getpid", lambda: 1001)
+
+    first = session_module.get_engine("postgresql+psycopg://user:pass@db/app")
+    second = session_module.get_engine("postgresql+psycopg://user:pass@db/app")
+
+    assert first is second
+    assert len(calls) == 1
+
+    monkeypatch.setattr(session_module.os, "getpid", lambda: 1002)
+    third = session_module.get_engine("postgresql+psycopg://user:pass@db/app")
+
+    assert third is not first
+    assert len(calls) == 2
+
+
+def test_auto_migrate_cached_per_pid(monkeypatch) -> None:
+    auto_migrate_module._auto_migrate_cached.cache_clear()
+    calls: list[tuple[str, object]] = []
+
+    monkeypatch.setattr(auto_migrate_module, "get_engine", lambda url: f"engine-for-{url}")
+    monkeypatch.setattr(
+        auto_migrate_module,
+        "auto_migrate_engine",
+        lambda engine: calls.append(("auto_migrate_engine", engine)),
+    )
+    monkeypatch.setattr(auto_migrate_module.os, "getpid", lambda: 2001)
+
+    auto_migrate_module.auto_migrate("postgresql+psycopg://user:pass@db/app")
+    auto_migrate_module.auto_migrate("postgresql+psycopg://user:pass@db/app")
+
+    assert calls == [("auto_migrate_engine", "engine-for-postgresql+psycopg://user:pass@db/app")]
+
+    monkeypatch.setattr(auto_migrate_module.os, "getpid", lambda: 2002)
+    auto_migrate_module.auto_migrate("postgresql+psycopg://user:pass@db/app")
+
+    assert calls == [
+        ("auto_migrate_engine", "engine-for-postgresql+psycopg://user:pass@db/app"),
+        ("auto_migrate_engine", "engine-for-postgresql+psycopg://user:pass@db/app"),
+    ]

tests/test_httpx_proxy.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from unittest import TestCase
+
+from videoroll.utils.httpx_proxy import (
+    HTTPX_SOCKS_SUPPORT_MISSING,
+    format_httpx_proxy_error,
+    uses_socks_proxy,
+)
+
+
+class HttpxProxyUtilsTests(TestCase):
+    def test_uses_socks_proxy_detects_socks_schemes(self) -> None:
+        self.assertTrue(uses_socks_proxy("socks5://127.0.0.1:1080"))
+        self.assertTrue(uses_socks_proxy("socks4://127.0.0.1:9050"))
+        self.assertFalse(uses_socks_proxy("http://127.0.0.1:7890"))
+
+    def test_format_httpx_proxy_error_rewrites_missing_socks_support(self) -> None:
+        error = RuntimeError(
+            "Using SOCKS proxy, but the 'socksio' package is not installed. "
+            "Make sure to install httpx using `pip install httpx[socks]`."
+        )
+
+        self.assertEqual(
+            format_httpx_proxy_error(error, proxy="socks5://127.0.0.1:1080"),
+            HTTPX_SOCKS_SUPPORT_MISSING,
+        )
+
+    def test_format_httpx_proxy_error_keeps_other_errors(self) -> None:
+        error = RuntimeError("connection timeout")
+
+        self.assertEqual(
+            format_httpx_proxy_error(error, proxy="http://127.0.0.1:7890"),
+            "connection timeout",
+        )

tests/test_resume_failed_recent.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+import uuid
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from fastapi import HTTPException
+
+from videoroll.apps.orchestrator_api import main
+from videoroll.db.models import SourceLicense, SourceType, Task, TaskStatus
+
+
+class _FakeQuery:
+    def __init__(self, *, all_result=None, count_result: int | None = None):
+        self._all_result = all_result
+        self._count_result = count_result
+
+    def filter(self, *_args, **_kwargs):
+        return self
+
+    def order_by(self, *_args, **_kwargs):
+        return self
+
+    def limit(self, *_args, **_kwargs):
+        return self
+
+    def all(self):
+        return self._all_result
+
+    def count(self):
+        if self._count_result is None:
+            raise AssertionError("count_result was not configured")
+        return self._count_result
+
+
+def _failed_task(*, source_type: SourceType, source_url: str | None = None) -> Task:
+    return Task(
+        id=uuid.uuid4(),
+        source_type=source_type,
+        source_url=source_url,
+        source_license=SourceLicense.own,
+        status=TaskStatus.failed,
+    )
+
+
+def test_resume_failed_recent_restarts_youtube_pipeline_when_no_subtitle_job(monkeypatch) -> None:
+    task = _failed_task(source_type=SourceType.youtube, source_url="https://www.youtube.com/watch?v=abc123")
+    db = MagicMock()
+    db.query.side_effect = [
+        _FakeQuery(all_result=[task]),
+        _FakeQuery(count_result=0),
+        _FakeQuery(count_result=0),
+    ]
+
+    set_created_by_calls: list[tuple[uuid.UUID, str]] = []
+
+    monkeypatch.setattr(main, "_build_auto_publish_after_render", lambda *args, **kwargs: {"publish": True})
+    monkeypatch.setattr(
+        main,
+        "_build_resume_subtitle_request",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            HTTPException(status_code=400, detail="no subtitle job found to resume")
+        ),
+    )
+    monkeypatch.setattr(main, "get_auto_profile", lambda _db: {"auto_publish": True})
+    monkeypatch.setattr(
+        main,
+        "_set_task_created_by",
+        lambda _settings, *, task_id, created_by: set_created_by_calls.append((task_id, created_by)),
+    )
+    monkeypatch.setattr(main, "_enqueue_auto_youtube_pipeline", lambda task_id, auto_publish: f"pipeline:{task_id}:{auto_publish}")
+
+    resp = main.resume_recent_failed_tasks(
+        window_hours=24,
+        limit=200,
+        settings=SimpleNamespace(),
+        db=db,
+        s3=MagicMock(),
+    )
+
+    assert resp.resumed_count == 1
+    assert resp.skipped_count == 0
+    assert resp.failed_count == 0
+    assert resp.results[0].task_id == task.id
+    assert resp.results[0].status == "queued"
+    assert "started auto_youtube pipeline" in str(resp.results[0].detail)
+    assert set_created_by_calls and set_created_by_calls[0][0] == task.id
+
+
+def test_resume_failed_recent_skips_non_youtube_task_without_subtitle_job(monkeypatch) -> None:
+    task = _failed_task(source_type=SourceType.local, source_url=None)
+    db = MagicMock()
+    db.query.side_effect = [
+        _FakeQuery(all_result=[task]),
+        _FakeQuery(count_result=0),
+        _FakeQuery(count_result=0),
+    ]
+
+    monkeypatch.setattr(main, "_build_auto_publish_after_render", lambda *args, **kwargs: {"publish": True})
+    monkeypatch.setattr(
+        main,
+        "_build_resume_subtitle_request",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            HTTPException(status_code=400, detail="no subtitle job found to resume")
+        ),
+    )
+
+    resp = main.resume_recent_failed_tasks(
+        window_hours=24,
+        limit=200,
+        settings=SimpleNamespace(),
+        db=db,
+        s3=MagicMock(),
+    )
+
+    assert resp.resumed_count == 0
+    assert resp.skipped_count == 1
+    assert resp.failed_count == 0
+    assert resp.results[0].task_id == task.id
+    assert resp.results[0].status == "skipped"
+    assert resp.results[0].detail == "no subtitle job found to resume"