From 26c0020a5cd72b90019b7bf16ccc62bcd60d122d Mon Sep 17 00:00:00 2001 From: mingyang <577523479@qq.com> Date: Tue, 3 Nov 2026 08:16:27 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A4=A7=E6=96=87=E4=BB=B6=E5=88=86?= =?UTF-8?q?=E7=89=87=E7=BA=A7=E6=96=AD=E7=82=B9=E7=BB=AD=E4=BC=A0=EF=BC=88?= =?UTF-8?q?=E4=BC=9A=E8=AF=9D=E8=90=BD=20SQLite=EF=BC=8Cjob=20=E5=B1=82?= =?UTF-8?q?=E6=B3=A8=E5=85=A5=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 探针阶段二已证明服务端认几小时前的 uniqueId/batchNo。这一版按 (b) 落地： Wopan 只懂协议，会话生命周期在 job 层。 分层 - Wopan.upload_parts(ses, info, fh, start, on_part)：从第 start 片接到末片， upload() 与续传共用同一个循环，避免两条路径悄悄分叉 - engine.upload_resumable()：建/复用/判废会话、按当前 token 重加密 fileInfo、 每 32 片刷一次 done；不认识 sqlite，SessionStore 由 runner/main 注入 - store.SessionStore：upload_sessions 单表，与 DB 同库另开连接 会话快照存整份 {uid, batch_no, info_plain, size, total, done, token_fp} - 只存 uniqueId 续不上：服务端还比 batchNo 与加密后的 fileInfo - 存 info_plain 而非密文：密文 key 由 access_token 派生，token 一换就解不开 - token_fp 不符 → 保守重开会话（对应探针里「token 绑定未知」那条未验证分支） 判废与收尾 - 键 = (dir_id, relpath, size, mtime_ns)，文件一改自动失效 - TTL 24h，懒清理（prune 在每次 sync/push 开头跑一次） - verify 分流：哈希/体积/未找到 → 删会话（已证伪）；复核时列目录抛异常 → 保留会话（transient，下次直接续，不用重传 1.2GiB） 并发 - tpbackup.db 是 rollback journal（非 WAL），写锁全库：不做多线程并发写， 每 32 片由 job 侧串行刷一次，锁只罩一次 UPSERT 开关 - 默认只对 ≥16MiB 的文件启用（TP_RESUME_MIN_SIZE 可调，TP_RESUME=0 关闭）； 小文件本来一片，存会话纯亏，也把 blast radius 压在这个阈值以上 验证 - 29 项单测全绿（新增 12 项：分片偏移/短读/会话往返/TTL/三件套沿用/判废） - 真机回归（容器内 /tmp 跑新代码，未动线上）：小文件 sync 2 个 + push 1 个 复核通过；24MB 文件人为断在第 3 片 → done=2 落盘 → 续传只补第 3 片， uniqueId 沿用，云端 sha256 与本地逐字一致 顺带修：part_count 的 part 默认值改为调用时取 PART（原写法定义时绑定， 改 PART 不生效，测试与将来做可配置分片都会踩）。 --- app/engine.py | 276 +++++++++++++++++++++++++++++++++++++------ app/main.py | 7 +- app/runner.py | 5 + app/store.py | 109 +++++++++++++++++ tests/test_pure.py | 5 +- tests/test_resume.py | 229 +++++++++++++++++++++++++++++++++++ 6 files changed, 594 insertions(+), 37 deletions(-) create mode 100644 tests/test_resume.py diff --git a/app/engine.py b/app/engine.py index 608bdde..86539c0 100644 --- a/app/engine.py +++ b/app/engine.py @@ -60,14 +60,18 @@ def _gai_v4(host, port, family=0, type=0, proto=0, flags=0): PART = 8 * 1024 * 1024 -def part_count(size, part=PART): +def part_count(size, part=None): """分片数，向上取整。 v0.2.0 及以前是 size // PART：20MB 会切成 8MB + 12MB，15.9MB 直接 整包单片发（total 算成 1，客户端自己认为「只有一片」）。服务端能收， 但末片比标准片还大，语义是错的——一旦服务端收紧 partSize 校验， 表现就是「小文件能传、大文件必挂」。这里只改片数，协议字段不动。 + + part 默认在**调用时**取模块常量，而不是定义时绑定：写成 part=PART 的话 + 改 PART 对已有函数无效，测试改不大、将来做可配置分片也会踩同一个坑。 """ + part = PART if part is None else part return max(1, -(-int(size) // part)) @@ -114,6 +118,46 @@ def new_unique_id(): for _ in range(6))) +# ── 断点续传的会话快照 ────────────────────────────────────── +# 只有「大文件」才值得续传：8MiB 以下的文件本来就一片，存会话只是多一次 +# 磁盘写。阈值可调，写死在代码里是为了默认行为可预测。 +RESUME_MIN_SIZE = int(os.environ.get("TP_RESUME_MIN_SIZE", str(16 * 1048576))) +RESUME_ENABLED = os.environ.get("TP_RESUME", "1").lower() not in ("0", "false", "no", "off") +# 每 32 片把 done 刷一次盘：8MiB × 32 = 256MiB，崩了最多重传这么多。 +SESSION_FLUSH_EVERY = 32 + + +def token_fingerprint(at): + """access_token 指纹。fileInfo 的加密 key 由它派生，token 一换 key 就换， + 旧会话的 fileInfo 再也复现不出来——所以要按指纹判会话还能不能用。""" + return hashlib.sha256((at or "").encode()).hexdigest()[:12] + + +def new_upload_session(dirid, name, size, batch_no=None): + """开一次上传会话，返回完整快照。续传时必须整份复用，不能只留 uniqueId。""" + plain = file_info_plain(dirid, name, size, batch_no=batch_no) + return {"dir_id": str(dirid), "name": name, "size": int(size), + "total": part_count(size), "uid": new_unique_id(), + "batch_no": plain["batchNo"], "info_plain": plain, "done": 0} + + +def should_resume(size, enabled=None, threshold=None): + on = RESUME_ENABLED if enabled is None else enabled + return bool(on) and int(size) >= (RESUME_MIN_SIZE if threshold is None else threshold) + + +def session_key(dir_id, rel_path, size, mtime_ns): + """会话键：(目标目录, 相对路径, 体积, mtime_ns)。 + + 文件一改，mtime_ns 和 size 至少一个变，旧会话自动失效——不需要额外的 + 「文件是否变了」判断。最坏是键碰撞（同体积同 mtime 的不同内容）， + 那也只是 verify 报错，不会静默写坏：最终有 SHA256 把关。 + 纯字符串运算，不碰 sqlite，所以留在 engine 这边。 + """ + raw = "%s\x00%s\x00%d\x00%d" % (dir_id, rel_path, int(size), int(mtime_ns)) + return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] + + # 恢复出来的文件归属：默认给 NAS 主用户（fnOS 的 1000:1000） RESTORE_UID = int(os.environ.get("TP_FILE_UID", "1000")) RESTORE_GID = int(os.environ.get("TP_FILE_GID", "1000")) @@ -293,36 +337,60 @@ def mkdir(self, parent, name, retries=5): raise last def upload(self, path, dirid, name=None, on_part=None): + """整文件上传（默认路径）。 + + 分片循环本身搬到了 upload_parts()：续传要从第 N 片接上，两条路径 + 必须共用同一个循环，否则「续传版」和「普通版」会悄悄分叉——探针 + 已经吃过一次这种亏（手抄字段测的是自己的实现）。 + """ name = name or os.path.basename(path) - size = os.path.getsize(path) - part = min(PART, size) or 1 - total = part_count(size) - info = enc_file_info(self.key, file_info_plain(dirid, name, size)) - uid = new_unique_id() - fid, done = "", 0 - last_err = None + ses = new_upload_session(dirid, name, os.path.getsize(path)) + info = enc_file_info(self.key, ses["info_plain"]) with open(path, "rb") as f: - for idx in range(1, total + 1): - cur = part if idx != total else size - done - blob = f.read(cur) - fields = self.part_fields(dirid, name, size, total, uid, info, idx, cur) - for attempt in range(3): - try: - r = self._upload_part(fields, name, blob) - if r.get("code") != "0000": - raise RuntimeError("code=%s msg=%s" % (r.get("code"), r.get("msg"))) - if (r.get("data") or {}).get("fid"): - fid = r["data"]["fid"] - last_err = None - break - except Exception as e: - last_err = e - time.sleep(2 * (attempt + 1)) - if last_err: - raise last_err - done += cur - if on_part: - on_part(idx, total) + return self.upload_parts(ses, info, f, 1, on_part) + + def upload_parts(self, ses, info, fh, start=1, on_part=None): + """发第 start..total 片，返回 fid（服务端最后一次给的那个）。 + + ses：一次上传会话的快照——uniqueId / batchNo / fileInfo 明文 / 体积 / + 总片数。三件套必须与首片完全一致，服务端才认这是同一次上传。 + fh ：已打开的文件句柄；start>1 时由本函数自己 seek，调用方不用管。 + """ + size = int(ses["size"]) + total = int(ses["total"]) + dirid, name, uid = ses["dir_id"], ses["name"], ses["uid"] + part = min(PART, size) or 1 + start = max(1, min(int(start), total + 1)) + done = (start - 1) * part + if start > 1: + fh.seek(done) # 前面的片服务端已经收下，不重传 + fid, last_err = "", None + for idx in range(start, total + 1): + cur = part if idx != total else size - done + blob = fh.read(cur) + if len(blob) != cur: + # 文件在传输中被改短了。继续发会让 partSize 与实际字节不符， + # 服务端可能收下一份拼不起来的东西，宁可当场失败。 + raise RuntimeError("本地读取不足：第 %d 片要 %d 字节，只读到 %d" + % (idx, cur, len(blob))) + fields = self.part_fields(dirid, name, size, total, uid, info, idx, cur) + for attempt in range(3): + try: + r = self._upload_part(fields, name, blob) + if r.get("code") != "0000": + raise RuntimeError("code=%s msg=%s" % (r.get("code"), r.get("msg"))) + if (r.get("data") or {}).get("fid"): + fid = r["data"]["fid"] + last_err = None + break + except Exception as e: + last_err = e + time.sleep(2 * (attempt + 1)) + if last_err: + raise last_err + done += cur + if on_part: + on_part(idx, total) return fid def part_fields(self, dirid, name, size, total, uid, info, idx, part_size): @@ -368,6 +436,74 @@ def _upload_part(self, fields, filename, blob): shutil.rmtree(d, ignore_errors=True) +# ── 带断点续传的上传（job 层，会话存储由调用方注入）────────── +def upload_resumable(w, sessions, skey, path, dirid, name=None, + log=lambda m: None, on_part=None): + """上传一个文件，能从上次中断的片号接上。 + + 分层：Wopan 只懂协议（upload_parts），会话的存取、判废、落盘全在这里， + Wopan 不知道 sqlite 的存在。会话存的是**整份快照**（uniqueId + batchNo + + fileInfo 明文），只存 uniqueId 续不上——服务端还比 batchNo 与加密后的 + fileInfo。 + + 成功时**不删会话**：要等 verify 通过才删。删早了，一旦哈希不符就只能 + 从零重传；留着才能按「证伪」处理。 + """ + name = name or os.path.basename(path) + st = os.stat(path) + size, mtime_ns = st.st_size, st.st_mtime_ns + fp = token_fingerprint(w.at) + + s = sessions.get(skey) + if s: + why = "" + if int(s["size"]) != size or int(s["mtime_ns"]) != mtime_ns: + why = "文件已变（体积或 mtime 动了）" + elif str(s["dir_id"]) != str(dirid): + why = "目标目录变了" + elif s.get("token_fp") and s["token_fp"] != fp: + # token 一换，fileInfo 的加密 key 也跟着换。虽然我们存的是明文、 + # 能重加密，但会话是不是绑旧 token 服务端没给答案——按保守策略 + # 重开一次，与探针报告里那条「未验证分支」保持一致。 + why = "token 已轮换（%s → %s）" % (s["token_fp"], fp) + if why: + log("放弃旧会话 %s：%s，重开一次上传" % (name, why)) + sessions.drop(skey) + s = None + + if not s: + s = new_upload_session(dirid, name, size) + s.update(skey=skey, mtime_ns=mtime_ns, token_fp=fp, done=0) + sessions.save(s) + start = 1 + else: + start = min(int(s["done"]) + 1, int(s["total"]) + 1) + if start > 1: + log("续传 %s：从第 %d/%d 片接上（会话更新于 %s）" % ( + name, start, s["total"], s.get("updated_at") or "-")) + + # 明文重加密：存的 info_plain，key 由当前 token 派生 + info = enc_file_info(w.key, s["info_plain"]) + + def prog(idx, total): + s["done"] = idx + if idx % SESSION_FLUSH_EVERY == 0 or idx == total: + sessions.save(s) + if on_part: + on_part(idx, total) + + try: + with open(path, "rb") as f: + fid = w.upload_parts(s, info, f, start, prog) + except Exception: + s["done"] = min(int(s.get("done") or 0), int(s["total"])) + sessions.save(s) # 保住已完成的进度，下次接着来 + raise + s["done"] = int(s["total"]) + sessions.save(s) + return fid + + # ── 目录链与扫描 ──────────────────────────────────────────── def ensure_chain(w, names, log=lambda m: None): cid = "0" @@ -665,7 +801,7 @@ def collect(w, src, cid, jobs, stats, exclude_dirs, exclude_files, rel="", log=l # ── 一次完整同步 ──────────────────────────────────────────── def sync(store, src, cloud_chain, excludes_dirs=None, excludes_files=None, dry_run=False, limit=0, workers=4, log=lambda m: None, - progress=None, should_stop=None, verify=True): + progress=None, should_stop=None, verify=True, sessions=None): """执行一次同步。 store : TokenStore 实例，负责 token 的读写与续期 @@ -676,6 +812,9 @@ def sync(store, src, cloud_chain, excludes_dirs=None, excludes_files=None, should_stop: 可选回调，返回 True 时中止（用于 Web 上点"停止"） verify : 上传后做内容复核（体积全比 + ≤VERIFY_SHA256_MAX 的文件比 SHA256）。 真实上传路径默认开启；dry-run 不经过上传，与本参数无关。 + sessions: 可选 SessionStore。给了且文件够大（TP_RESUME_MIN_SIZE，默认 + 16MiB）才走断点续传；不给则每个文件都是一次全新会话， + 行为与 v0.2.1 一致。 返回 dict(stats)；空文件（0 字节）不上传，单独计入 empty，不并入 skip。 """ result = {"ok": False, "reason": "", "todo": 0, "ok_n": 0, "fail": 0, @@ -767,6 +906,15 @@ def sync(store, src, cloud_chain, excludes_dirs=None, excludes_files=None, done_n, done_b, fail = 0, 0, 0 err_msgs, failed, succeeded = [], [], [] lock = threading.Lock() + skeys = {} # relp -> 会话键（只有走了续传的文件才有） + + if sessions is not None and RESUME_ENABLED: + try: + n = sessions.prune() + if n: + log("清理了 %d 条过期的上传会话" % n) + except Exception as e: + log("清理过期会话失败（不影响本次同步）: %s" % e) def work(j): nonlocal done_n, done_b, fail @@ -775,7 +923,15 @@ def work(j): return False t = time.time() try: - w.upload(p, cid, name) + # 大文件走会话续传，小文件照旧：8MiB 以下本来就只有一片， + # 为它存会话纯属多一次磁盘写，没有收益。 + if sessions is not None and should_resume(sz): + skey = session_key(cid, relp, sz, os.stat(p).st_mtime_ns) + upload_resumable(w, sessions, skey, p, cid, name, log=log) + with lock: + skeys[relp] = (skey, str(cid)) + else: + w.upload(p, cid, name) el = time.time() - t with lock: done_n += 1 @@ -823,6 +979,21 @@ def work(j): # 体积全比；云端返回了 sha256 且文件 ≤64MB 的再比内容哈希。 # 大文件只比体积是刻意的取舍——全量哈希会让 1GB 批次的 I/O 翻倍。 verify_bad, hashed = [], 0 + + def _drop_session(relp): + """复核判负 → 丢掉会话（已证伪，留着下次还会失败）。 + 拿不到校验数据（列目录抛异常）不走这里：那是 transient， + 留着下次直接续，不用把 1.2GiB 再传一遍。""" + v = skeys.pop(relp, None) + k = v[0] if v else None + if k and sessions is not None: + try: + sessions.drop(k) + except Exception as e: + log("删除上传会话失败（不影响结果）: %s" % e) + + unknown_dirs = set() # 列目录失败的目录：里面的会话一律保留 + if verify and succeeded: by_dir = {} for j in succeeded: @@ -833,14 +1004,17 @@ def work(j): for f in w.ls(cid2) if f.get("type") == 1} except Exception as e: log("复核时列目录失败（不影响已完成上传）: %s" % e) + unknown_dirs.add(str(cid2)) continue for (p2, _c, name2, sz2, relp2) in jlist: c = cloud.get(name2.lower()) if not c: verify_bad.append("%s 云端未找到" % relp2) + _drop_session(relp2) continue if str(c.get("size")) != str(sz2): verify_bad.append("%s 体积不符 本地=%s 云端=%s" % (relp2, sz2, c.get("size"))) + _drop_session(relp2) continue rh = (c.get("sha256") or "").lower() if rh and sz2 <= VERIFY_SHA256_MAX: @@ -848,12 +1022,21 @@ def work(j): hashed += 1 else: verify_bad.append("%s 内容哈希不符（传输损坏）" % relp2) + _drop_session(relp2) if verify_bad: log("❌ 复核未通过 %d 项：%s" % (len(verify_bad), "；".join(verify_bad[:5]))) else: log("✅ 复核通过：%d 个文件%s" % (done_n, ("，SHA256 一致 %d 个" % hashed) if hashed else "")) + # 还剩着的会话：已确认过的删掉，别在库里堆「早就传完了」的行。 + # 列目录失败的目录跳过——那批文件的状态我们根本没看到， + # 删了就等于把「下次直接续」的机会扔掉。 + for relp in list(skeys): + if skeys[relp][1] in unknown_dirs: + continue + _drop_session(relp) + all_errs = err_msgs + verify_bad result.update(ok=(not all_errs), ok_n=done_n, fail=fail + len(verify_bad), bytes=done_b, elapsed=el, @@ -898,12 +1081,13 @@ def _connect(store, log): def push(store, files, cloud_chain, keep=0, pattern="", dry_run=False, - log=lambda m: None, verify=True, protect=()): + log=lambda m: None, verify=True, protect=(), sessions=None): """把一批本地文件推送到云端目录（同名覆盖）。 files : [(本地绝对路径, 云端文件名或 None), ...] 比 sync() 少一个"扫描目录"的过程——宿主机脚本已经把包装好了， 这里只负责传输、校验、保留策略。凭证与重试逻辑与 sync() 共用。 + sessions : 可选 SessionStore，给了就对大文件启用断点续传（同 sync）。 """ result = {"ok": False, "reason": "", "todo": 0, "ok_n": 0, "fail": 0, "exist": 0, "skip": 0, "bytes": 0, "elapsed": 0, @@ -959,10 +1143,22 @@ def push(store, files, cloud_chain, keep=0, pattern="", dry_run=False, log("云端根目录 id=%s" % root_id) ok_n, done_b, errs, uploaded = 0, 0, [], [] + skeys = {} # 文件名 -> 会话键（仅走续传的文件） + if sessions is not None and RESUME_ENABLED: + try: + sessions.prune() + except Exception as e: + log("清理过期会话失败（不影响本次推送）: %s" % e) + for p, n, sz in items: t = time.time() try: - w.upload(p, root_id, n) + if sessions is not None and should_resume(sz): + skey = session_key(root_id, n, sz, os.stat(p).st_mtime_ns) + upload_resumable(w, sessions, skey, p, root_id, n, log=log) + skeys[n] = skey + else: + w.upload(p, root_id, n) el = time.time() - t ok_n += 1 done_b += sz @@ -974,6 +1170,7 @@ def push(store, files, cloud_chain, keep=0, pattern="", dry_run=False, log("❌ 上传失败 %s: %s" % (n, e)) # 复核：云端体积必须一致；云端若返回 sha256，再比内容哈希（更强） + ls_failed = False if verify and uploaded: try: cloud = {} @@ -1011,6 +1208,19 @@ def push(store, files, cloud_chain, keep=0, pattern="", dry_run=False, "，SHA256 一致 %d 个" % hashed if hashed else "")) except Exception as e: log("复核时列目录失败（不影响已完成的上传）: %s" % e) + ls_failed = True + + # 会话收尾：传成功的、以及复核明确判负的，都删；只有「没拿到校验数据」 + # （列目录抛异常）才留着——那是 transient，留着下次直接续，不用重传。 + if skeys: + if ls_failed: + log("复核数据不可用，保留 %d 个上传会话供下次续传" % len(skeys)) + else: + for n in list(skeys): + try: + sessions.drop(skeys.pop(n)) + except Exception as e: + log("删除上传会话失败（不影响结果）: %s" % e) # 空文件记失败：显式点名要推的文件没传上去，就算没成功。 # 放在 prune 之前，失败的一轮不做保留策略清理。 diff --git a/app/main.py b/app/main.py index 43ac665..c4e93ef 100644 --- a/app/main.py +++ b/app/main.py @@ -36,7 +36,7 @@ from apscheduler.triggers.cron import CronTrigger from .runner import Runner from .scheduler import Scheduler -from .store import DB, TokenStore +from .store import DB, SessionStore, TokenStore logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") @@ -58,6 +58,9 @@ os.makedirs(os.path.join(DATA, "logs"), exist_ok=True) db = DB(os.path.join(DATA, "tpbackup.db")) +# 断点续传的会话账本。sync 走 runner 注入，push 走这里注入—— +# 两处都注入，engine 就不用反向依赖 store。 +sessions = SessionStore(os.path.join(DATA, "tpbackup.db")) store = TokenStore(os.path.join(DATA, "state", "token.json"), openlist_db=OPENLIST_DB or None, sync_openlist=SYNC_OPENLIST, log=log.info) @@ -423,7 +426,7 @@ def logf(m): try: res = engine_push(store=store, files=files, cloud_chain=cloud, keep=d.keep, pattern=d.pattern, dry_run=d.dry_run, - log=logf, protect=PROTECT_FILES) + log=logf, protect=PROTECT_FILES, sessions=sessions) except Exception as e: res = {"ok": False, "reason": "exception", "detail": str(e)[:300], "todo": len(files), "ok_n": 0, "fail": len(files), diff --git a/app/runner.py b/app/runner.py index 013558a..2ce1c0b 100644 --- a/app/runner.py +++ b/app/runner.py @@ -9,6 +9,7 @@ from . import engine from .notify import kuma +from .store import SessionStore class RunState: @@ -43,6 +44,9 @@ class Runner: def __init__(self, db, store, logs_dir, on_change=None): self.db = db self.store = store + # 断点续传的账本：与 DB 同一个 sqlite 文件，单独连接。 + # 放在这里注入给 engine，engine 自己不认识 sqlite。 + self.sessions = SessionStore(db.path) self.logs_dir = logs_dir self.on_change = on_change self._states = {} @@ -142,6 +146,7 @@ def progress(p): excludes_dirs=excludes, excludes_files=excludes, dry_run=dry_run, workers=task.get("workers") or 4, log=log, progress=progress, should_stop=lambda: st.stop_flag, + sessions=self.sessions, ) if st.stop_flag: diff --git a/app/store.py b/app/store.py index 1499029..24f6fb7 100644 --- a/app/store.py +++ b/app/store.py @@ -195,6 +195,27 @@ def recover(self, w, prefer_db=True): log TEXT DEFAULT '' ); CREATE INDEX IF NOT EXISTS idx_runs_task ON runs(task_id, id DESC); + +-- 分片上传会话：断点续传的账本。 +-- 一行 = 一次「uniqueId + batchNo + fileInfo」的上传会话，必须整份留住： +-- 只存 uniqueId 续不上（服务端还比 batchNo 和加密后的 fileInfo）。 +-- info_plain 存明文、不存密文：密文是拿 access_token 派生的 key 加密的， +-- token 一换就再也解不出原来的 fileInfo，探针已经踩过这条路。 +CREATE TABLE IF NOT EXISTS upload_sessions ( + skey TEXT PRIMARY KEY, + dir_id TEXT NOT NULL, + name TEXT NOT NULL, + size INTEGER NOT NULL, + mtime_ns INTEGER NOT NULL, + total INTEGER NOT NULL, + done INTEGER DEFAULT 0, + uid TEXT NOT NULL, + batch_no TEXT NOT NULL, + info_plain TEXT NOT NULL, + token_fp TEXT DEFAULT '', + updated_at TEXT +); +CREATE INDEX IF NOT EXISTS idx_sessions_updated ON upload_sessions(updated_at); """ # 老库补列：SQLite 没有 IF NOT EXISTS 的 ADD COLUMN，只能逐列尝试 @@ -366,3 +387,91 @@ def vacuum(self): return True except Exception: return False + + +# ── 断点续传会话（job 层用，engine 不碰 sqlite）─────────────── +SESSION_TTL = 24 * 3600 # 秒。只证过 2h6m 的间隔，24h 是保守取值 + + +class SessionStore: + """分片上传会话的持久化。 + + 与 DB 共用同一个 sqlite 文件，但单独开连接。注意库是 rollback journal + （不是 WAL），写会锁全库：所以写入必须串行且短——_lk 只罩住一次 + UPDATE/INSERT，绝不跨网络调用持有。engine 侧每 32 片刷一次，就是这个原因。 + """ + + def __init__(self, path): + self.path = path + self._lk = threading.Lock() + os.makedirs(os.path.dirname(path), exist_ok=True) + with self._conn() as c: + c.executescript(SCHEMA) + + def _conn(self): + c = sqlite3.connect(self.path, timeout=20) + c.row_factory = sqlite3.Row + return c + + def get(self, skey, ttl=SESSION_TTL): + """取会话。超过 TTL 直接删掉并返回 None（懒清理，不另起定时器）。""" + with self._conn() as c: + r = c.execute("select * from upload_sessions where skey=?", + (skey,)).fetchone() + if not r: + return None + d = dict(r) + if ttl and d.get("updated_at"): + try: + age = time.time() - time.mktime( + time.strptime(d["updated_at"], "%Y-%m-%d %H:%M:%S")) + except Exception: + age = None + if age is not None and age > ttl: + self.drop(skey) + return None + try: + d["info_plain"] = json.loads(d["info_plain"]) + except Exception: + self.drop(skey) + return None + return d + + def save(self, d): + now = time.strftime("%F %T") + info = d["info_plain"] + if not isinstance(info, str): + info = json.dumps(info, ensure_ascii=False, separators=(",", ":")) + with self._lk, self._conn() as c: + c.execute("""insert into upload_sessions + (skey,dir_id,name,size,mtime_ns,total,done,uid,batch_no, + info_plain,token_fp,updated_at) + values(?,?,?,?,?,?,?,?,?,?,?,?) + on conflict(skey) do update set + dir_id=excluded.dir_id, name=excluded.name, + size=excluded.size, mtime_ns=excluded.mtime_ns, + total=excluded.total, done=excluded.done, + uid=excluded.uid, batch_no=excluded.batch_no, + info_plain=excluded.info_plain, + token_fp=excluded.token_fp, + updated_at=excluded.updated_at""", + (d["skey"], str(d["dir_id"]), d["name"], int(d["size"]), + int(d["mtime_ns"]), int(d["total"]), int(d.get("done") or 0), + d["uid"], d["batch_no"], info, + d.get("token_fp") or "", now)) + + def drop(self, skey): + with self._lk, self._conn() as c: + c.execute("delete from upload_sessions where skey=?", (skey,)) + + def prune(self, ttl=SESSION_TTL): + """清掉过期行。sync/push 开始时调一次即可，不必逐个定时。""" + with self._lk, self._conn() as c: + cur = c.execute( + "delete from upload_sessions where updated_at < datetime('now','localtime',?)", + ("-%d seconds" % int(ttl),)) + return cur.rowcount + + def count(self): + with self._conn() as c: + return c.execute("select count(*) from upload_sessions").fetchone()[0] diff --git a/tests/test_pure.py b/tests/test_pure.py index a83ae1e..d816391 100644 --- a/tests/test_pure.py +++ b/tests/test_pure.py @@ -163,7 +163,8 @@ def fake_sync(**kw): monkeypatch.setattr(runner_mod.engine, "sync", fake_sync) class FakeDB: - def __init__(self): + def __init__(self, path): + self.path = path # Runner 会拿它开续传会话表 self.finished = [] def start_run(self, tid, name): @@ -176,7 +177,7 @@ def prune_runs(self, *a, **k): pass logs = tmp_path / "logs" - r = runner_mod.Runner(FakeDB(), object(), str(logs)) + r = runner_mod.Runner(FakeDB(str(tmp_path / "tp.db")), object(), str(logs)) task = {"id": 1, "name": "t", "src": str(tmp_path), "cloud": '["A"]', "excludes": '["*.tmp", "#recycle"]', "workers": 2, "mode": "dir"} assert r.start(task) == 1 diff --git a/tests/test_resume.py b/tests/test_resume.py new file mode 100644 index 0000000..ac5f1c9 --- /dev/null +++ b/tests/test_resume.py @@ -0,0 +1,229 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""断点续传的纯逻辑单测：分片偏移 / 会话落盘 / 判废条件。 + +不碰网络：_upload_part 被替换成一个只记账的假实现，所以这里测的是 +「第几片、发多少字节、uniqueId 与 fileInfo 有没有换」这些真正容易错的地方。 +真机路径另由 tools/probe_resume.py 回归。 +""" +import os +import sqlite3 +import sys + +import pytest + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) + +from app import engine # noqa: E402 +from app.store import SessionStore # noqa: E402 + + +class FakeUpload(engine.Wopan): + """只替换 _upload_part 的 Wopan：不建连接、不发请求。 + + part_fields 用真的——字段构造是这次要保护的对象，不能用假的替掉。 + """ + + def __init__(self, token="T" * 8, fail_at=None): + self.at = token + self.key = b"0123456789abcdef" # 16 字节，够 aes_enc 用 + self.calls = [] + self.fail_at = fail_at + + def _upload_part(self, fields, filename, blob): + idx = int(fields["partIndex"]) + if self.fail_at and idx == self.fail_at: + raise RuntimeError("模拟断线@%d" % idx) + self.calls.append({"idx": idx, "n": len(blob), + "uid": fields["uniqueId"], + "info": fields["fileInfo"], + "at": fields["accessToken"]}) + return {"code": "0000", "data": {"fid": "F%d" % idx}} + + +@pytest.fixture +def small_part(monkeypatch): + """把分片大小临时调小，好在一屏内写完测试。 + + 顺带把重试的 sleep 掐掉：失败用例要跑满 3 次重试，真睡就是 12 秒。 + """ + monkeypatch.setattr(engine, "PART", 1000) + monkeypatch.setattr(engine.time, "sleep", lambda s: None) + return 1000 + + +def _file(tmp_path, size): + p = tmp_path / "blob.bin" + p.write_bytes(bytes((i * 37) % 256 for i in range(size))) + return str(p) + + +def _session(size, total, uid="U1", name="blob.bin", dirid="D1"): + ses = engine.new_upload_session(dirid, name, size) + ses["total"] = total # 小文件场景下由测试指定 + ses["uid"] = uid + return ses + + +def test_upload_parts_full_range(small_part, tmp_path): + """不续传时：1..total 全发，末片按剩余字节算。""" + p = _file(tmp_path, 5400) + ses = _session(5400, 6) + w = FakeUpload() + with open(p, "rb") as f: + fid = w.upload_parts(ses, "INFO", f, 1) + assert [c["idx"] for c in w.calls] == [1, 2, 3, 4, 5, 6] + assert [c["n"] for c in w.calls] == [1000, 1000, 1000, 1000, 1000, 400] + assert fid == "F6" + + +def test_upload_parts_resumes_and_seeks(small_part, tmp_path): + """从第 4 片续：文件指针必须落到 offset=3000，且只发 4..6。""" + p = _file(tmp_path, 5400) + ses = _session(5400, 6) + w = FakeUpload() + with open(p, "rb") as f: + fid = w.upload_parts(ses, "INFO", f, 4) + assert [c["idx"] for c in w.calls] == [4, 5, 6] + assert [c["n"] for c in w.calls] == [1000, 1000, 400] + # 末片大小 = size - (total-1)*PART，与整传一致 + assert sum(c["n"] for c in w.calls) == 5400 - 3000 + assert fid == "F6" + + +def test_upload_parts_start_beyond_total_is_noop(small_part, tmp_path): + """done==total 时（传完但没验过）再来一次：一片都不发，不报错。""" + p = _file(tmp_path, 5400) + ses = _session(5400, 6) + w = FakeUpload() + with open(p, "rb") as f: + fid = w.upload_parts(ses, "INFO", f, 7) + assert w.calls == [] + assert fid == "" + + +def test_upload_parts_short_read_raises(small_part, tmp_path): + """文件被截短：宁可当场失败，也不要发一份 partSize 不对的片。""" + p = _file(tmp_path, 5400) + ses = _session(9000, 9) # 声称 9000，实际只有 5400 + w = FakeUpload() + with open(p, "rb") as f: + with pytest.raises(RuntimeError, match="本地读取不足"): + w.upload_parts(ses, "INFO", f, 1) + + +def test_session_key_changes_with_mtime(): + a = engine.session_key("D1", "x/y.bin", 100, 111) + b = engine.session_key("D1", "x/y.bin", 100, 111) + c = engine.session_key("D1", "x/y.bin", 100, 112) + assert a == b + assert a != c + + +def test_token_fingerprint_stable_and_sensitive(): + assert engine.token_fingerprint("abc") == engine.token_fingerprint("abc") + assert engine.token_fingerprint("abc") != engine.token_fingerprint("abd") + + +def test_should_resume_threshold(monkeypatch): + assert engine.should_resume(1024, enabled=True, threshold=0) is True + assert engine.should_resume(engine.RESUME_MIN_SIZE - 1, + enabled=True, threshold=engine.RESUME_MIN_SIZE) is False + assert engine.should_resume(engine.RESUME_MIN_SIZE, + enabled=True, threshold=engine.RESUME_MIN_SIZE) is True + assert engine.should_resume(10 ** 9, enabled=False) is False + + +def test_session_store_roundtrip(tmp_path): + s = SessionStore(str(tmp_path / "t.db")) + ses = engine.new_upload_session("D1", "a.bin", 5000) + ses.update(skey="K1", mtime_ns=42, token_fp="fp1", done=7) + s.save(ses) + + got = s.get("K1") + assert got["done"] == 7 + assert got["uid"] == ses["uid"] + assert got["batch_no"] == ses["batch_no"] + # info_plain 必须回来是 dict：续传时要拿它按当前 key 重加密 + assert isinstance(got["info_plain"], dict) + assert got["info_plain"]["batchNo"] == ses["batch_no"] + assert got["token_fp"] == "fp1" + + s.drop("K1") + assert s.get("K1") is None + + +def test_session_store_ttl_drops_stale_row(tmp_path): + db = str(tmp_path / "t.db") + s = SessionStore(db) + ses = engine.new_upload_session("D1", "a.bin", 5000) + ses.update(skey="K2", mtime_ns=1, token_fp="fp", done=3) + s.save(ses) + # 手动把 updated_at 推到 2 天前 + c = sqlite3.connect(db) + c.execute("update upload_sessions set updated_at=datetime('now','localtime','-2 days')") + c.commit() + c.close() + + assert s.get("K2", ttl=24 * 3600) is None, "超过 TTL 的会话必须判废" + assert s.count() == 0, "判废时应顺手删掉，别留垃圾" + + +def _upload(sessions, path, dirid="D1", name="a.bin", w=None, skey="K9"): + w = w or FakeUpload() + return engine.upload_resumable(w, sessions, skey, path, dirid, name), w + + +def test_resumable_keeps_progress_then_continues(small_part, tmp_path): + """断在第 4 片：done 必须落成 3；下一次从第 4 片接上，且会话三件套不变。""" + p = _file(tmp_path, 5400) + s = SessionStore(str(tmp_path / "t.db")) + + # 第一次：第 4 片断线 + with pytest.raises(RuntimeError, match="模拟断线"): + _upload(s, p, w=FakeUpload(fail_at=4)) + row = s.get("K9") + assert row is not None, "失败后会话必须留着，否则下次只能从零重传" + assert row["done"] == 3 + first_uid, first_batch = row["uid"], row["batch_no"] + + # 第二次：应当从第 4 片接上 + _, w2 = _upload(s, p) + assert [c["idx"] for c in w2.calls] == [4, 5, 6] + assert all(c["uid"] == first_uid for c in w2.calls), "uniqueId 必须沿用" + assert all(c["info"] for c in w2.calls) + # 传给服务端的仍是同一份 fileInfo（同一 batchNo、同一 key） + assert len({c["info"] for c in w2.calls}) == 1 + assert s.get("K9")["batch_no"] == first_batch, "batchNo 必须沿用，否则就是另一次上传" + assert s.get("K9")["done"] == 6, "跑完后 done 必须等于 total" + + +def test_resumable_restarts_on_token_change(small_part, tmp_path): + """token 换过 → 按保守策略重开会话（新 uniqueId），不从旧片续。""" + p = _file(tmp_path, 5400) + s = SessionStore(str(tmp_path / "t.db")) + + with pytest.raises(RuntimeError): + _upload(s, p, w=FakeUpload(token="A" * 8, fail_at=3)) + old_uid = s.get("K9")["uid"] + + _, w2 = _upload(s, p, w=FakeUpload(token="B" * 8)) + assert s.get("K9")["uid"] != old_uid, "换 token 后必须是新会话" + assert [c["idx"] for c in w2.calls] == [1, 2, 3, 4, 5, 6], "新会话要从头传" + assert all(c["uid"] == s.get("K9")["uid"] for c in w2.calls) + + +def test_resumable_restarts_when_file_changes(small_part, tmp_path): + """文件 mtime 变了（内容可能也变了）→ 旧会话判废，从头传。""" + p = _file(tmp_path, 5400) + s = SessionStore(str(tmp_path / "t.db")) + with pytest.raises(RuntimeError): + _upload(s, p, w=FakeUpload(fail_at=5)) + old_uid = s.get("K9")["uid"] + + os.utime(p, ns=(0, 999_999_999_999)) # 改 mtime，内容不变 + _, w2 = _upload(s, p) + assert s.get("K9")["uid"] != old_uid + assert [c["idx"] for c in w2.calls] == [1, 2, 3, 4, 5, 6]