From 750d3371a517a0f424f0e6e63cb63ea6cac41d65 Mon Sep 17 00:00:00 2001 From: kelriclink Date: Tue, 30 Jun 2026 23:25:46 +0800 Subject: [PATCH] fix: harden publish retry and proxy checks --- pyproject.toml | 2 +- src/videoroll/apps/bilibili_publisher/main.py | 41 +++++- .../apps/bilibili_publisher/worker.py | 76 ++++++++++- src/videoroll/apps/orchestrator_api/main.py | 46 ++++++- src/videoroll/apps/subtitle_service/main.py | 5 +- src/videoroll/db/auto_migrate.py | 9 +- src/videoroll/db/session.py | 9 +- src/videoroll/utils/httpx_proxy.py | 26 ++++ src/web/src/pages/SettingsYouTubePage.tsx | 9 +- tests/test_bilibili_publish_idempotency.py | 54 ++++++++ tests/test_bilibili_publish_rate_limit.py | 26 +++- tests/test_db_session_pid_isolation.py | 55 ++++++++ tests/test_httpx_proxy.py | 35 +++++ tests/test_resume_failed_recent.py | 121 ++++++++++++++++++ 14 files changed, 498 insertions(+), 16 deletions(-) create mode 100644 src/videoroll/utils/httpx_proxy.py create mode 100644 tests/test_bilibili_publish_idempotency.py create mode 100644 tests/test_db_session_pid_isolation.py create mode 100644 tests/test_httpx_proxy.py create mode 100644 tests/test_resume_failed_recent.py diff --git a/pyproject.toml b/pyproject.toml index 455be07..ed9cd19 100644 --- a/pyproject.toml +++ b/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", diff --git a/src/videoroll/apps/bilibili_publisher/main.py b/src/videoroll/apps/bilibili_publisher/main.py index 8768686..03789bc 100644 --- a/src/videoroll/apps/bilibili_publisher/main.py +++ b/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={ diff --git a/src/videoroll/apps/bilibili_publisher/worker.py b/src/videoroll/apps/bilibili_publisher/worker.py index d10653e..081edc0 100644 --- a/src/videoroll/apps/bilibili_publisher/worker.py +++ b/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() diff --git a/src/videoroll/apps/orchestrator_api/main.py b/src/videoroll/apps/orchestrator_api/main.py index 06a5ee3..6f8ea09 100644 --- a/src/videoroll/apps/orchestrator_api/main.py +++ b/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 diff --git a/src/videoroll/apps/subtitle_service/main.py b/src/videoroll/apps/subtitle_service/main.py index 34a9283..c57b2d9 100644 --- a/src/videoroll/apps/subtitle_service/main.py +++ b/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), ) diff --git a/src/videoroll/db/auto_migrate.py b/src/videoroll/db/auto_migrate.py index 7739696..bf10abf 100644 --- a/src/videoroll/db/auto_migrate.py +++ b/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()) diff --git a/src/videoroll/db/session.py b/src/videoroll/db/session.py index 42c4c72..17e0687 100644 --- a/src/videoroll/db/session.py +++ b/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) diff --git a/src/videoroll/utils/httpx_proxy.py b/src/videoroll/utils/httpx_proxy.py new file mode 100644 index 0000000..f04bb56 --- /dev/null +++ b/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 diff --git a/src/web/src/pages/SettingsYouTubePage.tsx b/src/web/src/pages/SettingsYouTubePage.tsx index 0a2c802..5828b75 100644 --- a/src/web/src/pages/SettingsYouTubePage.tsx +++ b/src/web/src/pages/SettingsYouTubePage.tsx @@ -352,14 +352,19 @@ export default function SettingsYouTubePage() {
代理设置
- 例如：http://127.0.0.1:7890 / socks5://127.0.0.1:1080（留空=不使用代理） + 支持 http:// 和 socks5://，例如 + http://127.0.0.1:7890 / + socks5://127.0.0.1:1080（留空 = 不使用代理） +
+
+ SOCKS5 检测和抓取依赖后端安装 SOCKS 支持；当前生产镜像已内置该依赖。