fix: 分块大小减至5MB + Edge端chunk_size变更自动清理 + 断点续传防护
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后文件损坏
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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}
|
||||
|
||||
Reference in New Issue
Block a user