From 9b55b88edb598101ffb44c9d0003fefcf24c0715 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 13 Jul 2026 07:51:23 +0800 Subject: [PATCH] =?UTF-8?q?fix(worker):=20OSS=E4=B8=8A=E4=BC=A0=E5=B4=A9?= =?UTF-8?q?=E6=BA=83=E4=BF=AE=E5=A4=8D=20-=20connect=E8=B6=85=E6=97=B6=20+?= =?UTF-8?q?=20=E5=88=86=E7=89=87=E4=B8=8A=E4=BC=A0=20+=20=E6=80=BB?= =?UTF-8?q?=E8=B6=85=E6=97=B6=E4=BF=9D=E6=8A=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. oss_bucket() 增加 connect_timeout=10s，防止TCP握手挂死 2. upload_to_oss() 大文件(>=100MB)自动走分片上传，降低内存峰值减少OOM风险 3. upload_to_oss() 增加总超时保护(300s)，线程超时自动返回失败 4. 7个单元测试覆盖：超时配置、分片策略、超时保护、异常处理 --- apps/worker/video_processing/oss_helpers.py | 92 +++++++- tests/unit/test_oss_upload_crash_fix.py | 236 ++++++++++++++++++++ 2 files changed, 318 insertions(+), 10 deletions(-) create mode 100755 tests/unit/test_oss_upload_crash_fix.py diff --git a/apps/worker/video_processing/oss_helpers.py b/apps/worker/video_processing/oss_helpers.py index 0e988f6a3..8638e4567 100755 --- a/apps/worker/video_processing/oss_helpers.py +++ b/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 校验或临时下载）。 diff --git a/tests/unit/test_oss_upload_crash_fix.py b/tests/unit/test_oss_upload_crash_fix.py new file mode 100755 index 000000000..cb47cc8cd --- /dev/null +++ b/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() -- 2.54.0