diff --git a/fam-core/src/fam_core/dispatcher/dispatcher.py b/fam-core/src/fam_core/dispatcher/dispatcher.py index 051cb97..917e7f2 100644 --- a/fam-core/src/fam_core/dispatcher/dispatcher.py +++ b/fam-core/src/fam_core/dispatcher/dispatcher.py @@ -27,7 +27,7 @@ from .. import db_layer logger = setup_logger('fam-core.dispatcher') -CHUNK_SIZE = 10 * 1024 * 1024 # 10MB per chunk (smaller = fewer timeouts) +CHUNK_SIZE = 5 * 1024 * 1024 # 5MB per chunk (reliable at ~1Mbps upload) CHUNK_THRESHOLD = 20 * 1024 * 1024 # files > 20MB use chunked upload MAX_CHUNK_RETRIES = 3 @@ -145,8 +145,13 @@ 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 端已上传分块列表""" + def _query_uploaded_chunks(self, task_id, expected_total=None): + """查询 Edge 端已上传分块列表 + + 返回 (uploaded_set, edge_total_chunks)。 + 如果 expected_total 与 edge_total 不匹配(chunk_size 变更), + 返回空集让 Edge 自动清理旧分块。 + """ try: resp = requests.get( self.chunks_query_url, @@ -154,10 +159,17 @@ class Dispatcher: timeout=(30, 15) ) if resp.status_code == 200: - return set(resp.json().get('uploaded_chunks', [])) + data = resp.json() + uploaded = set(data.get('uploaded_chunks', [])) + edge_total = data.get('total_chunks', 0) + if expected_total and edge_total and edge_total != expected_total: + logger.warning(f"[task_id={task_id}] Edge total_chunks={edge_total} " + f"≠ expected={expected_total}(chunk_size 已变更),从头上传") + return set(), edge_total + return uploaded, edge_total except requests.RequestException: pass - return set() + return set(), 0 def _dispatch_chunked(self, task, payload, video_path, file_size): """大文件分块上传 + 断点续传 @@ -171,7 +183,7 @@ class Dispatcher: filename = os.path.basename(video_path) # 1. 查询已上传分块(断点续传) - uploaded_set = self._query_uploaded_chunks(task_id) + uploaded_set, edge_total = self._query_uploaded_chunks(task_id, expected_total=total_chunks) if uploaded_set: logger.info(f"[task_id={task_id}] 断点续传: 已有 {len(uploaded_set)}/{total_chunks} 块") @@ -198,7 +210,7 @@ class Dispatcher: 'filename': filename, }, files={'chunk': (f'chunk_{idx}', io.BytesIO(chunk_data))}, - timeout=(30, 120) + timeout=(60, 180) ) if cresp.status_code == 200: success = True @@ -211,7 +223,7 @@ class Dispatcher: time.sleep(5 * (attempt + 1)) if not success: - uploaded_now = self._query_uploaded_chunks(task_id) + 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) diff --git a/fam-edge/src/fam_edge/api_gateway/api_gateway.py b/fam-edge/src/fam_edge/api_gateway/api_gateway.py index ca20b42..7b5becd 100644 --- a/fam-edge/src/fam_edge/api_gateway/api_gateway.py +++ b/fam-edge/src/fam_edge/api_gateway/api_gateway.py @@ -168,11 +168,26 @@ def upload_chunk(): chunk_path = os.path.join(d, f"chunk_{chunk_index:04d}") try: + # 检查 total_chunks 是否变化(chunk_size 变更导致),自动清理旧分块 + meta_path = os.path.join(d, "meta.json") + if total_chunks and os.path.exists(meta_path): + import json + try: + with open(meta_path) as mf: + old_meta = json.load(mf) + if old_meta.get('total_chunks') and old_meta['total_chunks'] != total_chunks: + logger.warning(f"[task_id={task_id}] total_chunks 变更 " + f"({old_meta['total_chunks']}→{total_chunks}),清理旧分块") + for fn in os.listdir(d): + if fn.startswith('chunk_'): + os.remove(os.path.join(d, fn)) + except (json.JSONDecodeError, IOError): + pass + chunk_file.save(chunk_path) size_kb = os.path.getsize(chunk_path) / 1024 - # 写元数据(首次上传时) - meta_path = os.path.join(d, "meta.json") + # 写/更新元数据 if not os.path.exists(meta_path): import json meta = {"filename": filename, "total_chunks": total_chunks}