26 · 重试与执行租约：旧 Worker 恢复后还能提交吗#

小陈的导出读完第一批时，队列进度写入失败。系统自动再试两次，最后显示失败；小陈点击重试后终于完成。可假如旧执行只是卡住，并没有真正退出，它晚回来的一份 CSV 能不能盖掉新结果？

本章把“同轮自动再试”“用户新开一轮”“旧执行失去提交资格”分开推演。重试是允许重新做，租约是限制谁还可以提交，两者不是一个概念。

1. 先给同一任务的三个身份命名#

Redis jobId 由 taskJobId(T, run) 生成，形如 T-0。同一轮自动重试仍是这个 job；用户手动重试后是 T-1。页面显示第 run+1 轮，数据库 run 从 0 开始。

attempts 又是另一个数，记录当前队列执行提供的 attemptsMade+1。它不等于所有历史轮次累计执行次数，不能拿它做完整审计计数。

2. 自动重试的策略到底配置在哪#

打开 queue.ts 的 defaultJobOptions，再跳到 MAX_TASK_ATTEMPTS：

// 源码节选：createTaskQueue
attempts: MAX_TASK_ATTEMPTS,
backoff: { type: 'exponential', delay: 2000 },
removeOnComplete: { age: 86400, count: 1000 },
removeOnFail: { age: 604800, count: 1000 },

MAX_TASK_ATTEMPTS=3，包含首次，所以正常临时错误最多再试两次，退避基准约为 2 秒、4 秒。它不是每两秒固定重试，也不是失败任务永不清理。age、count 是 BullMQ 队列记录保留策略，和 CSV 保存 24 小时是不同生命周期。

processor.ts 的 catch 根据 TaskRejected 与 attemptsMade+1 决定是否 final：

权限变化、无效筛选、超过导出上限等不会靠再等两秒自行变合法。TaskRejected 是当前代码的分类合同，不意味着世界上永远无法修复；用户调整条件通常需要创建新任务。

3. 沿一次失败逐字段推演#

假设 T / run=0，测试让读取中第一次 updateProgress 抛错，三次都如此：

这张表表示在保存 CSV 之前失败。不能把它套到“数据库已保存结果，最后向队列汇报 100 失败”的窗口；后者账本已经 COMPLETED，catch 的 RUNNING 条件匹配 0 行后直接返回，不改账本，也不再抛这次队列汇报错误。

进度先写数据库，再调用 job.updateProgress。后一步失败可能使用户短暂看到部分进度；那不是已经保存了部分 CSV。下一次认领 progress 会重置为 0，也不承诺进度永远单调上升。

4. 租约怎样阻止旧结果覆盖#

认领阶段生成 randomUUID，保存为 executionId。后续进度写入带相同条件：

// 源码节选：processor.ts
const lease = { id: task.id, run: task.run, executionId };
const updated = await db.backgroundTask.updateMany({
  where: { ...lease, status: 'RUNNING' },
  data: { progress: value },
});
if (!updated.count) throw new LeaseLost();

可以把 executionId 类比成“本次提交的通行证”，但它是内部数据库条件，不是用户认证令牌。

推演接手：A 认领 E1 后卡住；恢复流程清 E1，B 认领为 E2；A 醒来带 E1 更新进度，匹配 0 行，因此不能继续以旧身份提交。最终 CSV 写入也带 lease+RUNNING，旧结果无法无条件覆盖新结果。

导出最终 updateMany 现在也检查 completed.count !== 1，匹配 0 行就抛 LeaseLost。外层在处理其他错误前识别它并返回，不调用最后的 job.updateProgress(100)。因此“失去租约”不是“当前业务永久无法完成”，旧执行者不能用 UnrecoverableError 要求队列结束。

还有一条容易漏掉的路径：A 已失去租约，随后读取成员得到 TaskRejected，或读取期间发生普通网络错误。catch 会尝试按 E1 把任务置 FAILED / WAITING；如果更新 0 行，同样直接返回。否则旧错误虽然没改数据库，仍可能触发队列的失败处理。源码把“能否报告这次失败”也绑定到租约：

// 源码节选：错误状态仍按 lease + RUNNING 更新。
const failed = await db.backgroundTask.updateMany({
  where: { ...lease, status: 'RUNNING' },
  data: { status: final ? 'FAILED' : 'WAITING', executionId: null, lastError },
});
if (!failed.count) return;

片段中的 lastError 代表源码生成的固定可读提示，完整代码还区分永久拒绝与重试错误。匹配 1 行才继续抛出对应错误，让 BullMQ 决定同轮重试或结束。

导出并发用例 分别暂停 A 在“读取后、写进度前”和“最终授权后、写结果前”，再让 B 换成 E2。测试先断言 A 返回后整行任务仍等于 E2 接手快照，A 没有报告 100；再释放 B，并用不同标题验证保存的是 B 的 CSV。另有旧权限拒绝晚到的用例，验证错误上报也遵守租约。

注意 processTask 返回也可能代表跳过旧轮次或旧租约。真正完成仍以当前账本的状态、正文和期限为准，不能把一次 Promise resolve 当作“这名执行者已生成文件”的证明。

5. 重复认领与提醒副作用的并发边界#

认领条件允许 WAITING 或 RUNNING。它没有声明“只有空租约才能认领”，也没有在每次认领时检查租约到期。正常消费由 BullMQ 锁协调；发生重复调用或恢复时，新认领可以换掉旧 executionId。

因此此处“租约”重点是数据库提交资格，不是阻止所有重复 CPU / 查询工作。两份计算可能都进行过，但旧资格无法通过相应条件写入。不能据此声称端到端严格只执行一次。

提醒的难点是：团队锁保护工单、成员与提醒事务，但认领在锁外。A 在锁内读过 E1 后，B 仍能将 RUNNING 的 executionId 换成 E2。前面查过资格，不能替代提交时再次确认资格。

旧实现末尾虽然按 E1 条件更新，但没有检查更新行数。通知唯一约束挡住重复行，工单 version 与系统事件却可能再增加一次。现在 processor 把提醒任务的完成更新作为事务提交门：

const completed = await tx.backgroundTask.updateMany({
  where: { ...lease, status: 'RUNNING' },
  data: {
    status: 'COMPLETED',
    progress: 100,
    resultSummary,
    executionId: null,
  },
});
if (completed.count !== 1) throw new LeaseLost();

可以把它理解成收银结账：购物单已算好，但付款资格失效，就不能只留下商品出库记录。这里的“撤销”由 PostgreSQL 事务完成，通知、工单版本、系统事件和任务完成属于同一笔提交。

为什么不能在事务里直接 return？return 表示事务正常结束，前面的写入会提交；必须抛错越过事务边界，让 $transaction 拒绝后才在外层识别 LeaseLost 并返回。它也不是 TaskRejected：新执行者已经接手，旧执行者不能把任务标为 FAILED，不能要求队列永久停止。旧执行者也不会调用最后的 job.updateProgress(100)。

如果 A 先成功完成条件更新，任务行的写锁会持有到提交。B 的认领更新需要等待；醒来后看到 COMPLETED，条件不匹配，直接退出。若恢复扫描先清空租约、改成 WAITING，A 同样匹配 0 行并回滚，等待之后重新投递。

这里保证的是这条提醒数据库链路在这些交错下原子提交，允许重复计算。它没有把 Redis 和 PostgreSQL 变成一个事务，也没有提供外部邮件的严格一次发送保证。导出与提醒现在都检查末尾条件写入的行数，但导出主要防止虚假报告完成，提醒还要撤销同一事务前面已经写入的三项副作用。

6. 手动重试为什么不直接复用旧 job#

从 TasksWorkspace 的 retry.mutate，追到 任务服务 的 retry：先查成员，再按团队和可访问范围查任务，只允许 FAILED；工作人员创建的团队范围导出若已降为客户则拒绝。

更新时 run+1，置 WAITING，progress / attempts 归零，清 lastError、executionId、dispatchedAt、正文、摘要和过期时间，runAt 设现在。task.id、ownerId、filters 与原始范围保留。

旧消息带 run=0，新的账本是 run=1，processor 开头就返回；新的 T-1 也不会被 T-0 失败记录的固定 ID 去重挡住。页面提示“已受理重试”，仍不是文件已经生成。

原筛选超过上限时，重试不会自动把 filters 缩小；必须新建更窄范围任务。失去权限也不会因为点击重试恢复。

7. 卡住恢复与最后汇报失败怎样排查#

dispatcher 将超过 2 分钟没推进 updatedAt 的 RUNNING 改回 WAITING、清租约和投递时间。阈值基于账本更新时间，不是独立的每任务租约到期字段；导出批次进度会推进它。到阈值还要等扫描成功走到恢复逻辑。

启动脚本 进入 runWorkerProcess，再通过 createTaskWorker 安装与测试相同的处理器和监听器。stalled 表示 Redis 执行锁丢失；failed 表示 BullMQ 记录了一次失败。两个事件都没有失败那次执行的 executionId，不能拿 taskId + run 代替租约。

为什么 failed 回调现在只记日志#

旧代码在失败耗尽时，用 taskId、run 和 RUNNING / WAITING 将账本改 FAILED、清 executionId。这比处理器的错误路径宽：同一个 run 可以发生 E1 → E2 接手，迟到的旧回调会把 E2 一起删掉。即使在回调里“先查询当前 executionId”，读到的也可能正是 E2，并不能证明它属于失败的 E1。

现在 failed 回调只输出 task.failed。exhausted 表示尝试预算已耗尽，或错误属于 BullMQ.UnrecoverableError（包括永久业务拒绝与 stalled 超限）；不再用错误文本包含 stalled 猜业务类别。终态由有依据的两条路径决定：

代价是：极端故障下页面可能暂时显示 RUNNING，需等两分钟阈值及成功扫描后才收敛。不能为了即时变红，让无租约的回调去覆盖新执行者。普通权限拒绝、导出上限或正常三次失败仍由处理器及时写入，不需要一律等两分钟。

六条真实 Worker 测试怎样接上此前的证据#

worker.integration.test.ts 连接真实 PostgreSQL、Redis，调用正式 createTaskWorker，复用 processTask 与正式事件监听器；它没有手工 emit stalled / failed，也没有直接调用私有检测方法。

1. 导出与提醒各一条 stalled 接手：A 在真实数据库读取后暂停，关闭 A 的自动续锁，锁时长设为 1000ms；B 用 100ms 检测间隔等待锁自然过期。收到真实 stalled 并确认 B 已认领 E2 后，释放 A 并注入旧读取错误。A 按 E1 写错误匹配 0 行，返回后账本整行仍等于 E2 快照。再让 B 执行，核对 Redis completed / progress=100 与数据库 COMPLETED；导出检查正文、摘要和期限，提醒检查通知 1 条、系统事件 1 条、version 恰好加 1。
2. 旧 failed 回调迟到：A 的读取失败，且写 WAITING 暂时失败，真实 BullMQ 因 attempts=1 进入 failed。测试只移除本例这个队列记录，模拟记录丢失；将本任务 updatedAt 调旧后调用真正 dispatcher，两轮扫描恢复并补投同 run。B 认领 E2 后，再放行旧回调可能发出的 SQL，账本必须保持 E2，最终正确完成。修复后回调根本不发这条 SQL；临时恢复旧实现，会观察到 E2 被改成 FAILED、executionId=null，测试能检出。
3. 正常临时失败重试：首次真实读取错误，同一真实 Worker 经 30ms 退避再次处理，数据库与队列 attemptsMade 最终均为 2，结果完成，第一次 failed 日志 exhausted=false。
4. 永久拒绝：无效筛选使 processor 立即写 FAILED 并抛 UnrecoverableError；真实队列也 FAILED，attemptsMade=1、exhausted=true，不靠事件回调代写账本。
5. stalled 超过上限：B 的 maxStalledCount=0，真实检测后队列失败；账本先仍 RUNNING。测试把 updatedAt 调旧，第一轮扫描变 WAITING、清租约，第二轮核对 Redis failed 后变 FAILED。A 恢复并带旧错误返回，不改变最终整行账本。

第 1 项参数化展开为两条，共六条测试。1000ms 锁、100ms 检测、30ms 退避只用于加速隔离实验，不修改正式 Worker 默认锁参数或正式 2000ms 退避。同步靠事件和 Promise 关卡，8 秒上限只用于发现同步点未到达，不靠 sleep 排列执行顺序。

这是进程内真实 BullMQ Worker 的消费与恢复集成，不启动完整 Web，也不执行启动脚本的心跳、信号退出或容器守护。调旧 updatedAt 是压缩恢复时钟，不声称真的等待了两分钟；移除单个测试 job 是模拟记录丢失，不等于验证 Redis 整库丢失或实际网络断线恢复。运行命令与隔离清理见第 27 章；第 30 章另有真实子进程的心跳、SIGTERM 排空与超时退出恢复测试。

故障 A：旧执行者返回，却没有报告 100。 查 taskId、run 和当前状态，确认是否有新认领。进度与提交处失去租约会安静退出，不能据此判断新任务失败；不要把旧 E1 写回数据库“恢复权限”。

故障 B：队列报错，页面却已完成。 检查 resultCsv、摘要和期限。业务可能已经提交，只在最后 job.updateProgress(100) 出错；处理器会保留结果并返回，后续队列通信仍可能另报错，catch 与失败回调都不会按正常条件覆盖已完成账本。重新执行开头也会跳过 COMPLETED。

8. 动手运行可重复的接手实验#

按第 27 章准备独立库后运行：

TEST_DATABASE_URL=postgresql://teamdesk:teamdesk@localhost:5433/teamdesk_stage5_test \
pnpm exec vitest run tests/tasks.integration.test.ts \
-t '提醒并发接手|提醒写入事件后故障|中途进度写入故障|手动重试' --reporter=verbose

现有测试 手工传 attemptsMade=0、1、2，并使 updateProgress 抛错，回查 WAITING / FAILED、attempts 和 resultCsv=null；再调用服务 retry，成功 execute 后断言 COMPLETED。这里没有真实等待退避，它验证 processor 的业务状态处理。

手动重试用例调用旧 run=0，检查任务仍 WAITING，再执行新轮并查 run=1。另有三条真实数据库用例专门补上提醒的证据：

1. 新执行者认领：Prisma $extends 在 A 的租约查询返回后暂停，在 B 的认领 UPDATE 真正提交后暂停。释放 A，先检查通知 0、系统事件 0、version 未变，任务完整保持 B 的 RUNNING 租约；再释放 B，检查通知 1、事件 1、version 恰好加 1、任务 COMPLETED / 100 / executionId=null。
2. 恢复扫描清除租约：在相同暂停点按本任务 ID 模拟扫描的状态更新。A 返回后仍为 WAITING，无副作用；之后正常执行只产生一份业务结果。这不等于已经测试 dispatcher 的两分钟计时与 Redis 补投。
3. 事件写入后故障：在事务中执行真实事件 INSERT 后抛错，验证通知、事件和 version 全回滚，任务 WAITING 且租约清空；第二次执行才得到唯一一份结果，并核对 attempts=2。

这些用例不使用 sleep 排列先后。Promise 关卡确定顺序，4 秒超时只用于发现同步点没有到达；finally 无论断言成功与否都释放关卡并等待执行者退出，避免残留占锁事务。每个并发用例最后再顺序执行一次，验证已完成任务不会追加副作用。

练习 A：怎样让“旧执行者不能覆盖”成为确定性测试#

练习 B：为什么 FAILED 后不能只把 status 改回 WAITING#

练习 C：attempts=3 是否意味着整个任务一生只运行三次#

链路：队列调度某轮 → 认领新 executionId → 按租约推进进度与结果 → 临时错误同轮退避，明确拒绝结束 → FAILED 经授权开启新 run → 旧轮或旧租约不能沿原条件提交。