apps/worker/video_processing/oss_helpers.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 import hashlib
 import logging
 import os
+import threading
 from pathlib import Path
 from typing import Optional
 from urllib.parse import urlparse
@@ -17,6 +18,13 @@ import oss2
 
 logger = logging.getLogger(__name__)
 
+# OSS 上传配置
+OSS_CONNECT_TIMEOUT = 10  # 连接超时（秒），防止 TCP 握手挂死
+OSS_UPLOAD_TOTAL_TIMEOUT = 300  # 单文件上传总超时（秒），防止网络慢时无限卡住
+OSS_MULTIPART_THRESHOLD = 100 * 1024 * 1024  # 分片上传阈值：100MB 以上走分片
+OSS_PART_SIZE = 8 * 1024 * 1024  # 分片大小：8MB
+OSS_MULTIPART_NUM_THREADS = 3  # 分片上传并发数
+
 
 # ── OSS 配置 ──────────────────────────────────────────────────────────────────
 
@@ -43,6 +51,9 @@ def oss_bucket() -> oss2.Bucket | None:
     P0-2 修复：endpoint 不带 scheme 时自动补 https:// 前缀，
     确保 sign_url 等依赖 scheme 的方法返回 HTTPS URL。
 
+    P0-staging 修复：增加 connect_timeout=10s，防止网络抖动时
+    TCP 握手阶段无限挂死，导致 worker 进程卡死。
+
     Returns:
         oss2.Bucket 实例，配置缺失时返回 None。
     """
@@ -53,7 +64,12 @@ def oss_bucket() -> oss2.Bucket | None:
     # endpoint 无 scheme 时补 https://，与 API 端 storage.py 保持一致
     if not endpoint.startswith(("http://", "https://")):
         endpoint = f"https://{endpoint}"
-    return oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
+    return oss2.Bucket(
+        oss2.Auth(access_key_id, access_key_secret),
+        endpoint,
+        bucket_name,
+        connect_timeout=OSS_CONNECT_TIMEOUT,
+    )
 
 
 def normalize_storage_key(storage_key_or_url: str) -> str:
@@ -96,6 +112,9 @@ def download_asset(asset_storage_key: str, local_path: Path) -> bool:
 def upload_to_oss(local_path: Path, storage_key: str) -> str | None:
     """上传文件到 OSS，返回公开 URL。
 
+    大文件（>100MB）自动走分片上传，降低内存峰值，减少 OOM 风险。
+    上传加总超时保护（默认 300s），防止网络异常时无限挂死。
+
     Args:
         local_path: 本地文件路径
         storage_key: 目标存储键
@@ -106,18 +125,71 @@ def upload_to_oss(local_path: Path, storage_key: str) -> str | None:
     bucket = oss_bucket()
     if bucket is None:
         return None
-    try:
-        bucket.put_object_from_file(storage_key, str(local_path))
-        settings = oss_settings()
-        if settings:
-            _, _, endpoint, bucket_name = settings
-            endpoint_clean = endpoint.replace("https://", "").replace("http://", "")
-            return f"https://{bucket_name}.{endpoint_clean}/{storage_key}"
+
+    result: dict = {"url": None, "error": None, "file_size": 0}
+    done = threading.Event()
+
+    def _do_upload():
+        try:
+            # 尝试获取文件大小，用于分片判断和日志；stat 失败时 fallback 走普通上传
+            try:
+                file_size = local_path.stat().st_size
+                result["file_size"] = file_size
+                use_multipart = file_size >= OSS_MULTIPART_THRESHOLD
+            except OSError:
+                use_multipart = False
+                file_size = 0
+
+            if use_multipart:
+                # 分片上传：降低内存峰值，每片 8MB，3 线程并发
+                logger.info(
+                    "大文件分片上传: storage_key=%s, size=%.1fMB, part_size=%dMB, threads=%d",
+                    storage_key[:80],
+                    file_size / 1024 / 1024,
+                    OSS_PART_SIZE // 1024 // 1024,
+                    OSS_MULTIPART_NUM_THREADS,
+                )
+                oss2.resumable_upload(
+                    bucket,
+                    storage_key,
+                    str(local_path),
+                    multipart_threshold=OSS_MULTIPART_THRESHOLD,
+                    part_size=OSS_PART_SIZE,
+                    num_threads=OSS_MULTIPART_NUM_THREADS,
+                )
+            else:
+                bucket.put_object_from_file(storage_key, str(local_path))
+
+            # 构造返回 URL
+            settings = oss_settings()
+            if settings:
+                _, _, endpoint, bucket_name = settings
+                endpoint_clean = endpoint.replace("https://", "").replace("http://", "")
+                result["url"] = f"https://{bucket_name}.{endpoint_clean}/{storage_key}"
+        except Exception as e:
+            result["error"] = e
+            logger.exception("上传 OSS 失败: %s", storage_key)
+        finally:
+            done.set()
+
+    upload_thread = threading.Thread(target=_do_upload, daemon=True)
+    upload_thread.start()
+    finished = done.wait(timeout=OSS_UPLOAD_TOTAL_TIMEOUT)
+
+    if not finished:
+        logger.error(
+            "OSS 上传超时（%.0fs），强制中止: storage_key=%s, size=%.1fMB",
+            OSS_UPLOAD_TOTAL_TIMEOUT,
+            storage_key[:80],
+            result["file_size"] / 1024 / 1024 if result["file_size"] else 0,
+        )
         return None
-    except Exception:
-        logger.exception("上传 OSS 失败: %s", storage_key)
+
+    if result["error"]:
         return None
 
+    return result["url"]
+
 
 def get_signed_download_url(storage_key_or_url: str, expires_seconds: int = 3600) -> str | None:
     """生成预签名下载 URL（用于私有 bucket 的 URL 校验或临时下载）。

tests/unit/test_oss_upload_crash_fix.py

@@ -0,0 +1,236 @@
+"""P0-staging：OSS 上传崩溃修复测试.
+
+测试：
+1. oss_bucket() 传递 connect_timeout 参数
+2. upload_to_oss() 小文件走 put_object_from_file，大文件走分片上传
+3. upload_to_oss() 超时保护（超过总超时返回 None）
+4. upload_to_oss() 异常时返回 None
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import time
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# ── oss_bucket connect_timeout 测试 ───────────────────────────────────────────
+
+
+class TestOSSBucketConnectTimeout:
+    """测试 oss_bucket() 传递 connect_timeout 参数."""
+
+    def test_oss_bucket_has_connect_timeout(self):
+        """oss_bucket 应传递 connect_timeout=10s 参数."""
+        from video_processing.oss_helpers import oss_bucket
+
+        mock_bucket_instance = MagicMock()
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
+        ):
+            bucket = oss_bucket()
+
+            assert bucket is mock_bucket_instance
+            # 验证 connect_timeout 关键字参数
+            call_kwargs = mock_bucket_cls.call_args[1]
+            assert "connect_timeout" in call_kwargs, "oss_bucket 应传递 connect_timeout 参数"
+            assert (
+                call_kwargs["connect_timeout"] == 10
+            ), f"connect_timeout 应为 10，实际为 {call_kwargs['connect_timeout']}"
+
+    def test_oss_bucket_no_config_returns_none(self):
+        """OSS 配置缺失时返回 None."""
+        from video_processing.oss_helpers import oss_bucket
+
+        with patch.dict(os.environ, {}, clear=True):
+            bucket = oss_bucket()
+            assert bucket is None
+
+
+# ── upload_to_oss 分片上传测试 ────────────────────────────────────────────────
+
+
+class TestUploadToOSSMultipart:
+    """测试 upload_to_oss() 根据文件大小选择上传方式."""
+
+    def _create_temp_file(self, size_bytes: int) -> Path:
+        """创建指定大小的临时文件."""
+        tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+        tmp.write(b"x" * size_bytes)
+        tmp.close()
+        return Path(tmp.name)
+
+    def test_small_file_uses_put_object(self):
+        """小文件（<100MB）走 put_object_from_file."""
+        from video_processing.oss_helpers import upload_to_oss
+
+        small_file = self._create_temp_file(10 * 1024 * 1024)  # 10MB
+        try:
+            mock_bucket = MagicMock()
+
+            with (
+                patch.dict(
+                    os.environ,
+                    {
+                        "OSS_ACCESS_KEY_ID": "test-key",
+                        "OSS_ACCESS_KEY_SECRET": "test-secret",
+                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                        "OSS_BUCKET_NAME": "test-bucket",
+                    },
+                ),
+                patch("video_processing.oss_helpers.oss2.Auth"),
+                patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
+                patch("video_processing.oss_helpers.oss2.resumable_upload") as mock_resumable,
+            ):
+                url = upload_to_oss(small_file, "test/small.mp4")
+
+                # 验证调用了 put_object_from_file
+                mock_bucket.put_object_from_file.assert_called_once()
+                # 验证没调用分片上传
+                mock_resumable.assert_not_called()
+                # 验证返回 URL
+                assert url == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/test/small.mp4"
+        finally:
+            small_file.unlink()
+
+    def test_large_file_uses_resumable_upload(self):
+        """大文件（>=100MB）走 resumable_upload 分片上传."""
+        from video_processing.oss_helpers import upload_to_oss
+
+        large_file = self._create_temp_file(100 * 1024 * 1024)  # 100MB
+        try:
+            mock_bucket = MagicMock()
+
+            with (
+                patch.dict(
+                    os.environ,
+                    {
+                        "OSS_ACCESS_KEY_ID": "test-key",
+                        "OSS_ACCESS_KEY_SECRET": "test-secret",
+                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                        "OSS_BUCKET_NAME": "test-bucket",
+                    },
+                ),
+                patch("video_processing.oss_helpers.oss2.Auth"),
+                patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
+                patch("video_processing.oss_helpers.oss2.resumable_upload") as mock_resumable,
+            ):
+                url = upload_to_oss(large_file, "test/large.mp4")
+
+                # 验证调用了分片上传
+                mock_resumable.assert_called_once()
+                # 验证没调用 put_object_from_file
+                mock_bucket.put_object_from_file.assert_not_called()
+                # 验证分片参数
+                call_kwargs = mock_resumable.call_args[1]
+                assert call_kwargs["multipart_threshold"] == 100 * 1024 * 1024
+                assert call_kwargs["part_size"] == 8 * 1024 * 1024
+                assert call_kwargs["num_threads"] == 3
+                # 验证返回 URL
+                assert url == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/test/large.mp4"
+        finally:
+            large_file.unlink()
+
+
+# ── upload_to_oss 超时测试 ────────────────────────────────────────────────────
+
+
+class TestUploadToOSSTimeout:
+    """测试 upload_to_oss() 超时保护."""
+
+    def test_upload_timeout_returns_none(self):
+        """上传超过总超时时返回 None."""
+        from video_processing.oss_helpers import upload_to_oss
+
+        small_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+        small_file.write(b"x" * 1024)  # 1KB
+        small_file.close()
+        file_path = Path(small_file.name)
+
+        def slow_upload(*args, **kwargs):
+            """模拟慢速上传，超过超时时间."""
+            time.sleep(2)
+
+        mock_bucket = MagicMock()
+        mock_bucket.put_object_from_file.side_effect = slow_upload
+
+        try:
+            with (
+                patch.dict(
+                    os.environ,
+                    {
+                        "OSS_ACCESS_KEY_ID": "test-key",
+                        "OSS_ACCESS_KEY_SECRET": "test-secret",
+                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                        "OSS_BUCKET_NAME": "test-bucket",
+                    },
+                ),
+                patch("video_processing.oss_helpers.oss2.Auth"),
+                patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
+                patch("video_processing.oss_helpers.OSS_UPLOAD_TOTAL_TIMEOUT", 1),  # 1秒超时
+            ):
+                url = upload_to_oss(file_path, "test/slow.mp4")
+                # 超时应返回 None
+                assert url is None, "上传超时应返回 None"
+        finally:
+            file_path.unlink()
+
+    def test_upload_exception_returns_none(self):
+        """上传异常时返回 None."""
+        from video_processing.oss_helpers import upload_to_oss
+
+        small_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+        small_file.write(b"x" * 1024)
+        small_file.close()
+        file_path = Path(small_file.name)
+
+        mock_bucket = MagicMock()
+        mock_bucket.put_object_from_file.side_effect = RuntimeError("Network error")
+
+        try:
+            with (
+                patch.dict(
+                    os.environ,
+                    {
+                        "OSS_ACCESS_KEY_ID": "test-key",
+                        "OSS_ACCESS_KEY_SECRET": "test-secret",
+                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                        "OSS_BUCKET_NAME": "test-bucket",
+                    },
+                ),
+                patch("video_processing.oss_helpers.oss2.Auth"),
+                patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
+            ):
+                url = upload_to_oss(file_path, "test/error.mp4")
+                assert url is None, "上传异常应返回 None"
+        finally:
+            file_path.unlink()
+
+    def test_upload_no_bucket_returns_none(self):
+        """OSS 未配置时返回 None."""
+        from video_processing.oss_helpers import upload_to_oss
+
+        small_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+        small_file.write(b"x" * 1024)
+        small_file.close()
+        file_path = Path(small_file.name)
+
+        try:
+            with patch.dict(os.environ, {}, clear=True):
+                url = upload_to_oss(file_path, "test/noconfig.mp4")
+                assert url is None
+        finally:
+            file_path.unlink()