From d6054de06082f3c9068f26673204fe87b9a19940 Mon Sep 17 00:00:00 2001
From: ericwyuan <ericwyuan@example.com>
Date: Thu, 20 Aug 2026 13:55:59 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20=E5=88=86=E5=9D=97=E5=A4=A7=E5=B0=8F?=
 =?UTF-8?q?=E5=87=8F=E8=87=B310MB=20+=20=E4=B8=8A=E4=BC=A0=E5=A4=B1?=
 =?UTF-8?q?=E8=B4=A5=E7=A1=AE=E8=AE=A4=E6=9C=BA=E5=88=B6=20+=20=E9=87=8D?=
 =?UTF-8?q?=E8=AF=95=E5=89=8D=E7=BD=AE=E6=A3=80=E6=9F=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- CHUNK_SIZE 20MB→10MB，CHUNK_THRESHOLD 50MB→20MB，降低单次超时概率
- 新增 _query_uploaded_chunks 独立方法，connect timeout 10s→30s
- 分块上传失败后查询 Edge 确认是否实际收到，避免响应丢失导致不必要重试
- 补全 uploaded_set.add(idx) 防止重复上传已成功分块
- _poll_once 中前置 retry_count 检查，超限任务直接标记 FAILED
---
 .../src/fam_core/dispatcher/dispatcher.py     | 55 ++++++++++++-------
 1 file changed, 36 insertions(+), 19 deletions(-)

diff --git a/fam-core/src/fam_core/dispatcher/dispatcher.py b/fam-core/src/fam_core/dispatcher/dispatcher.py
index 4c88409..051cb97 100644
--- a/fam-core/src/fam_core/dispatcher/dispatcher.py
+++ b/fam-core/src/fam_core/dispatcher/dispatcher.py
@@ -27,8 +27,8 @@ from .. import db_layer
 
 logger = setup_logger('fam-core.dispatcher')
 
-CHUNK_SIZE = 20 * 1024 * 1024        # 20MB per chunk
-CHUNK_THRESHOLD = 50 * 1024 * 1024  # files > 50MB use chunked upload
+CHUNK_SIZE = 10 * 1024 * 1024        # 10MB per chunk (smaller = fewer timeouts)
+CHUNK_THRESHOLD = 20 * 1024 * 1024  # files > 20MB use chunked upload
 MAX_CHUNK_RETRIES = 3
 
 
@@ -145,11 +145,25 @@ class Dispatcher:
         logger.error(f"[task_id={task_id}] Edge 返回异常状态码: {resp.status_code}")
         self._schedule_retry(task)
 
+    def _query_uploaded_chunks(self, task_id):
+        """查询 Edge 端已上传分块列表"""
+        try:
+            resp = requests.get(
+                self.chunks_query_url,
+                params={'task_id': task_id},
+                timeout=(30, 15)
+            )
+            if resp.status_code == 200:
+                return set(resp.json().get('uploaded_chunks', []))
+        except requests.RequestException:
+            pass
+        return set()
+
     def _dispatch_chunked(self, task, payload, video_path, file_size):
         """大文件分块上传 + 断点续传
 
         1. 查询 Edge 端已上传分块（断点续传）
-        2. 上传缺失分块（每块最多重试 3 次）
+        2. 上传缺失分块（每块最多重试 3 次，失败后查询 Edge 确认是否实际收到）
         3. 全部分块上传后调用 /assemble 合并入队
         """
         task_id = task['task_id']
@@ -157,20 +171,9 @@ class Dispatcher:
         filename = os.path.basename(video_path)
 
         # 1. 查询已上传分块（断点续传）
-        uploaded_set = set()
-        try:
-            resp = requests.get(
-                self.chunks_query_url,
-                params={'task_id': task_id},
-                timeout=(10, 15)
-            )
-            if resp.status_code == 200:
-                data = resp.json()
-                uploaded_set = set(data.get('uploaded_chunks', []))
-                if uploaded_set:
-                    logger.info(f"[task_id={task_id}] 断点续传: 已有 {len(uploaded_set)}/{total_chunks} 块")
-        except requests.RequestException as e:
-            logger.warning(f"[task_id={task_id}] 查询已上传分块失败（将从头上传）: {e}")
+        uploaded_set = self._query_uploaded_chunks(task_id)
+        if uploaded_set:
+            logger.info(f"[task_id={task_id}] 断点续传: 已有 {len(uploaded_set)}/{total_chunks} 块")
 
         # 2. 上传缺失分块
         try:
@@ -195,7 +198,7 @@ class Dispatcher:
                                     'filename': filename,
                                 },
                                 files={'chunk': (f'chunk_{idx}', io.BytesIO(chunk_data))},
-                                timeout=(10, 120)
+                                timeout=(30, 120)
                             )
                             if cresp.status_code == 200:
                                 success = True
@@ -208,10 +211,16 @@ class Dispatcher:
                             time.sleep(5 * (attempt + 1))
 
                     if not success:
-                        logger.error(f"[task_id={task_id}] 分块 {idx} 超过最大重试次数，安排文件级重试")
+                        uploaded_now = self._query_uploaded_chunks(task_id)
+                        if idx in uploaded_now:
+                            logger.info(f"[task_id={task_id}] 分块 {idx} 虽超时但 Edge 已收到，继续下一块")
+                            uploaded_set.add(idx)
+                            continue
+                        logger.error(f"[task_id={task_id}] 分块 {idx} 确认未收到，安排文件级重试")
                         self._schedule_retry(task)
                         return
 
+                    uploaded_set.add(idx)
                     if (idx + 1) % 5 == 0 or idx == total_chunks - 1:
                         logger.info(f"[task_id={task_id}] 分块进度: {idx + 1}/{total_chunks}")
         except IOError as e:
@@ -278,6 +287,14 @@ class Dispatcher:
 
         tasks = db_layer.get_pending_tasks(limit=1)
         for task in tasks:
+            if task['retry_count'] >= task['max_retries']:
+                db_layer.update_task_status(
+                    task['task_id'], 'FAILED',
+                    error_message=f"超过最大重试次数 {task['max_retries']}",
+                    failure_stage='callback'
+                )
+                logger.warning(f"[task_id={task['task_id']}] retry_count={task['retry_count']} >= max_retries={task['max_retries']}，标记 FAILED")
+                continue
             if self._should_retry(task):
                 try:
                     self._dispatch_one(task)