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)