批量文档处理中断后，别急着点“重新开始”

假设你正在处理一批合同：先解析 PDF，再翻译，最后导出交付文件。进度跑到一半，页面报了超时。你最想知道的并不是“该等几秒重试”，而是： 刚才那份文档，到底处理过没有？ 如果服务端已经完成工作，只是响应没传回来，再提交一次可能生成第二个任务。反过来，如果你只看见一个输出文件就跳过，也可能把写到一半的残缺文件当成成品。真正可靠的续跑，第一步不是循环，而是把“已完成”“明确失败”和“结果未知”分开。 先记住一条规则：超时后先核对，不是先重发 请求发出与本地记账之间，总有一个无法靠本地数据库完全覆盖的窗口。把数据库事务写得再严密，也不能让另一个服务的工作跟着本地回滚。所以，我们先承认这个窗口，再让恢复流程有据可查。 看到的现象 此刻真正知道什么 正确的下一步 接口超时，没拿到任务号 只知道响应没收到 按持久化请求键查询；查不到也先保留未知状态 查到任务仍在运行 服务端已经接单 等待同一任务，按服务方要求退避查询 查到任务成功 服务端已有结果 拉取产物，校验后落库，再标记完成 本地结果已保存 当前修订已有可用结果 核对输入与参数，命中后直接复用 这张表不是某个产品接口的保证。真实服务可能有任务查询延迟、幂等键有效期或记录保留期。能否重试，最终取决于那个接口明确承诺了什么；尤其不要把一次“未找到”理解成“绝对没有执行”。 响应丢失后，任务走向哪里？ 离线教学示意 · 不发送请求，不产生费用 1 响应丢失 2 重启核对 3 再次运行 查不到任务？ 本地状态 unknown 提交次数 1 是否允许盲目重发 不允许 服务端已经完成处理，但成功响应没有送达。本地只能记为“结果未知”，不能据此判断任务失败。 下一步：保留请求标识，先查询原任务。 全屏 用两个小账本，把丢失的响应找回来 下面的实验不接网络，也不做真实 PDF 解析或翻译。它把“转成大写”当作处理动作，用两个磁盘 SQLite 数据库分别代表客户端账本和演示服务的任务账本。这样，注意力就能放在恢复流程，而不是模型、网络或积分上。 我们故意让服务端先保存结果，再抛出超时。客户端保存 unknown ，关闭并重新打开本地数据库，然后查询原任务并找回结果。这是 连接重开后的恢复测试 ，不是断电、进程强杀或分布式并发测试。 保存下面的完整代码为 resume_demo.py ，运行 python resume_demo.py 。仅用 Python 标准库；本次实测环境为 Python 3.13.11 / SQLite 3.51.0。演示数据库位于临时目录，程序结束后自动清理，不会修改你的业务数据库。 """Offline, single-worker reconciliation demo. No HTTP, translation or billing.""" import hashlib import json from pathlib import Path import sqlite3 import tempfile def task_key(raw, options): # The example accepts string-valued options, not arbitrary JSON types. assert all(isinstance(k, str) and isinstance(v, str) for k, v in options.items()) config = json.dumps(options, sort_keys=True, separators=(",", ":")) return hashlib.sha256(raw + b"\0" + config.encode()).hexdigest() class DemoProvider: """Pretend remote service: durable key lookup, immediate completion.""" def __init__(self, path): self.db = sqlite3.connect(path) self.db.execute("CREATE TABLE IF NOT EXISTS remote " "(key TEXT PRIMARY KEY, result TEXT NOT NULL)") self.db.commit() self.submit_calls = 0 def submit(self, key, raw, lose_reply=False): self.submit_calls += 1 # Idempotency is an explicit promise of THIS FAKE provider. with self.db: self.db.execute("INSERT OR IGNORE INTO remote VALUES (?, ?)", (key, raw.decode().upper())) if lose_reply: raise TimeoutError("accepted, but response lost") return self.lookup(key) def lookup(self, key): row = self.db.execute("SELECT result FROM remote WHERE key=?", (key,)).fetchone() return None if row is None else row[0] def resume(db, provider, raw, options, lose_reply=False): key = task_key(raw, options) row = db.execute("SELECT state, result FROM jobs WHERE key=?", (key,)).fetchone() if row and row[0] == "done": return "cached", row[1] if row: # Uncertain requests are queried, never blindly submitted again. result = provider.lookup(key) if result is None: return "needs_review", None outcome = "recovered" else: # Commit the intent BEFORE the external side effect. with db: db.execute("INSERT INTO jobs VALUES (?, 'unknown', NULL)", (key,)) try: result = provider.submit(key, raw, lose_reply) except TimeoutError: return "unknown", None outcome = "created" with db: db.execute("UPDATE jobs SET state='done', result=? WHERE key=?", (result, key)) return outcome, result def main(): options = {"engine": "demo-v1", "target": "en"} with tempfile.TemporaryDirectory(prefix="doc-resume-") as tmp: root = Path(tmp) provider = DemoProvider(root / "provider.sqlite") db = sqlite3.connect(root / "jobs.sqlite") db.execute("CREATE TABLE jobs " "(key TEXT PRIMARY KEY, state TEXT NOT NULL, result TEXT)") db.commit() first = resume(db, provider, b"invoice", options, lose_reply=True) assert first == ("unknown", None) print("1", first) db.close() # Reopen the on-disk ledger; this is not a process-kill test. db = sqlite3.connect(root / "jobs.sqlite") recovered = resume(db, provider, b"invoice", options) assert recovered == ("recovered", "INVOICE") print("2", recovered) cached = resume(db, provider, b"invoice", options) assert cached == ("cached", "INVOICE") print("3", cached) assert provider.submit_calls == 1 print("submit_calls=1 (the recovery did not resubmit)") # Crash after local intent, before the provider accepts anything. missing = task_key(b"not-sent", options) with db: db.execute("INSERT INTO jobs VALUES (?, 'unknown', NULL)", (missing,)) assert resume(db, provider, b"not-sent", options) == ("needs_review", None) assert provider.submit_calls == 1 print("missing remote task -> needs_review; submit_calls=1") assert task_key(b"invoice", options) != task_key(b"revised", options) assert task_key(b"invoice", options) != task_key( b"invoice", {**options, "target": "ja"}) assert task_key(b"invoice", options) == task_key( b"invoice", {"target": "en", "engine": "demo-v1"}) print("PASS content / target / option-order checks") db.close() provider.db.close() if __name__ == "__main__": main() 运行结果说明了什么 1 ('unknown', None) 2 ('recovered', 'INVOICE') 3 ('cached', 'INVOICE') submit_calls=1 (the recovery did not resubmit) missing remote task -> needs_review; submit_calls=1 PASS content / target / option-order checks 第一次调用已经在演示服务里写好了结果，但客户端只收到超时。第二次返回 recovered ：结果来自对原任务的查询，不是再次提交。第三次返回 cached ：本地已经确认完成，直接复用。输出中的 submit_calls=1 统计的是演示服务的提交方法调用次数，不是实际扣费次数。 最后一个分支更重要：只有本地意图，服务端却查不到任务。示例不会自动“补发”，而是返回 needs_review 。这会牺牲一点自动化程度，却不会把无法确认的状态伪装成安全重试。 读代码时，别漏掉这四个细节 一、写意图在先，发请求在后。 本地先提交 unknown 记录。即使随后中断，恢复程序也知道这里有一次需要核对的尝试。这个顺序不保证任务一定送达；因此“写了意图但尚未发出”仍要有独立的恢复策略。 二、同一份任务，必须有稳定的身份。 示例用输入字节和排序后的字符串参数计算摘要。文件名不在键里：改名不该强迫重做，内容变化才应该。真实文档处理还应包含目标语言、引擎版本、OCR 模式、术语表修订等会改变结果的参数；规范化协议固定以后，不要悄悄变更默认值。 三、数据库唯一键，不是远程调用保险。 INSERT OR IGNORE 只影响它所在的数据库。这里演示服务自身用它实现稳定键去重；客户端的恢复路径则直接查询，没有再次调用 submit 。把一条唯一约束放进自己的缓存表，不能让远端接口自动具备幂等能力。 四、完成状态要对应可验证的产物。 示例把一个短字符串直接保存在数据库里，便于复现。真实 PDF、UDOC 或 HTML 通常存放在对象存储：先校验文件可读性、大小与摘要，再把对象键和修订信息提交到任务记录；不要把临时下载链接本身当作永久结果。 把示例接进真实文档流程，还差哪些东西？ 这份代码刻意只做单工作者演示，没有任务认领锁，也没有租约、超时回收和指数退避。两个工作者同时处理同一个输入，会触发竞争条件；不要把它直接当生产队列。工程落地时，要把“谁有权推进这个任务”与“这个任务是否已经完成”分别设计。 接外部服务时，至少问清四件事：能否按你生成的请求键查询；重复键且参数相同时会发生什么；记录和幂等键保留多久；查询失败、未找到与任务失败是否能明确区分。AWS 的幂等接口文档展示了一种具体契约，但它不能替其他服务背书。 放到 OmniDoc 相关的解析、翻译与原生文档工作流中，也应保留同样的交付账本：输入摘要、处理参数、远端任务号、产物对象键、校验结果和用量凭据。本文没有调用 OmniDoc 翻译接口，不证明任何接口具备相同的幂等保证；对接时应按实际 API 文档逐项确认。 交付前，做一次“断在中间”的验收 正常跑通只能证明顺风时能用。至少再演练：提交前中断、响应丢失、查询暂时不可用、结果下载到一半、落库前中断，以及两个工作者撞上同一任务。每个场景都应能回答：现在有哪些证据，下一步允许做什么，谁来处理长期未知状态。 好的续跑，不是把所有失败都重试一遍，而是让每一次继续都有依据。 缓存负责少做重复工作，任务账本负责解释当前状态，服务端契约决定能否安全重发。这三件事分清了，批量文档处理才不会越“自动化”越难追查。 核对资料： Python sqlite3 文档 · SQLite 冲突处理 · AWS 幂等请求契约示例 产品入口： OmniDoc （产品链接，不代表本文对接口能力或费用的承诺）。