From 5915cf4be334a7929095eff0aba6c8ceddbccb0a Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Thu, 20 Aug 2026 14:18:29 +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=B35MB=20+=20Edge=E7=AB=AFchunk=5Fsize=E5=8F=98?= =?UTF-8?q?=E6=9B=B4=E8=87=AA=E5=8A=A8=E6=B8=85=E7=90=86=20+=20=E6=96=AD?= =?UTF-8?q?=E7=82=B9=E7=BB=AD=E4=BC=A0=E9=98=B2=E6=8A=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NAS dispatcher: - CHUNK_SIZE 10MB→5MB（~1Mbps上行带宽下可靠传输） - chunk上传timeout (30,120)→(60,180)（增加连接和读取余量） - _query_uploaded_chunks 返回 (set, edge_total) 元组 - expected_total != edge_total 时跳过断点续传（防止chunk_size变更导致文件损坏） Edge api_gateway: - upload_chunk 检测 total_chunks 变更，自动清理旧分块 - 防止不同chunk_size的旧分块与新分块混合导致assemble后文件损坏 --- .../src/fam_core/dispatcher/dispatcher.py | 28 +++++++++++++------ .../src/fam_edge/api_gateway/api_gateway.py | 19 +++++++++++-- 2 files changed, 37 insertions(+), 10 deletions(-) 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}