fix(worker): 显式重投无退避且 start 载体无 resume 语义——5 次预算在 ~60ms 内烧光（相邻间隔 1ms），每轮从首个节点整条重放并用 context={} 覆写盘上 checkpoint

Start in _requeue_retryable_failure in plaita/server/flow_worker.py and the except NodeExecutionRetryableError branch of run(), where a retry is enqueued and then acked with no delay. Read tests/unit/test_worker_node_retry.py to see the existing fakeredis fixtures. The fix needs a backoff chosen by design, so confirm the approach with a maintainer first. Done means retry intervals grow with attempt, the fifth failure still reaches a terminal error, and the test stubs the clock or sleep instead of really sleeping.

优先级 P1 · 依赖：无

基线 HEAD = 65f48f2

Summary

run() 对可重试节点失败的处理（b975109）改成「立即入队同体新副本 + ack 原条」，由此删掉了这条路径上唯一的节奏来源（旧语义是「不 ack 留 pending，等 claim_min_idle_ms（默认 60s）被 XCLAIM 回收」），却没有补任何退避；同时 start 载体的重投没有 resume 语义——每一轮都从首个节点整条重跑，并用 context={} 覆写盘上的 running 行。

1）重投本体：先 enqueue 再 ack，中间没有任何延迟。函数头（plaita/server/flow_worker.py:2548-2549）：

    def _requeue_retryable_failure(self, task: Any, queue: Any) -> bool:
        """节点可重试失败：**显式重投**一份同体新副本（ack 原条）。

函数体关键段（plaita/server/flow_worker.py:2571-2577）：

        try:
            queue.enqueue(dict(body))
        except Exception as e:  # noqa: BLE001 — 重投失败退化为 reclaim（绝不丢消息）
            logger.warning("任务 %s 显式重投失败（退化为留 pending 回收）: %s", mid, e)
            return False
        try:
            queue.ack(mid)

反证——函数体 2548-2581 内没有任何退避设施：

$ sed -n "2548,2581p" plaita/server/flow_worker.py | grep -n "sleep\|backoff\|delay"
（零命中，grep 退出码 1）
$ grep -n "sleep\|backoff" plaita/server/flow_worker.py
2926:            time.sleep(0.05)

（2926 属 wait_for_idle 的轮询等待，不在重投路径上。队列层同样没有延迟投递设施，唯一的 sleep 是连接重建退避：）

$ grep -n "sleep\|backoff\|delay" plaita/server/task_queue.py
189:            time.sleep(RECONNECT_BACKOFF_SECONDS)

2）调用点（plaita/server/flow_worker.py:2799-2806）——ack 原条后，新副本 delivery 归 1，队列层的 max_deliveries 死信出口不再作用于重投，预算只由 worker 侧计数键判定：

                if self._requeue_retryable_failure(task, queue):
                    acked = True
                    logger.warning(
                        "任务 %s 节点执行失败：已显式重投同体新副本（原 delivery=%s）: %s",
                        task.message_id,
                        task.delivery_count,
                        exc,
                    )

3）预算与生产默认值。预算门（plaita/server/flow_worker.py:656-657）：

        attempt = self._record_node_retry(execution_id)
        if attempt >= self._node_retry_budget():

预算取值（plaita/server/flow_worker.py:564-568）与队列默认常量（plaita/server/task_queue.py:18-19）：

        configured = getattr(self, "_max_deliveries", None)
        try:
            return max(1, int(configured)) if configured else DEFAULT_MAX_DELIVERIES
        except (TypeError, ValueError):
            return DEFAULT_MAX_DELIVERIES

DEFAULT_CLAIM_MIN_IDLE_MS = 60_000
DEFAULT_MAX_DELIVERIES = 5

CLI 默认值即走这两条常量（plaita/server/flow_worker.py:3127-3129、:3133-3135）：

    parser.add_argument("--claim-min-idle-ms", type=int,
                      default=int(os.environ.get("PLAITA_CLAIM_MIN_IDLE_MS", str(DEFAULT_CLAIM_MIN_IDLE_MS))),
                      help="pending 消息最短空闲毫秒数后才可被其他 consumer 回收")

    parser.add_argument("--max-deliveries", type=int,
                      default=int(os.environ.get("PLAITA_MAX_DELIVERIES", str(DEFAULT_MAX_DELIVERIES))),
                      help="任务最大投递次数，超过后写入 DLQ 并 ack")

4）消费循环也不提供任何节奏：新副本 XADD 后，下一次 read 立即命中（plaita/server/flow_worker.py:2722-2730）：

        while self._running and not self._drain_event.is_set():
            self._sweep_residue_if_due(queue)
            # 分片阻塞读取（2026-09 分布式评审 P2-2）：XREADGROUP 的
            # BLOCK 无法被信号中断出循环，整块 10s 会让 SIGTERM 后的
            # worker 继续抢任务最长 10s。切成 ≤1s 的窗口，停机延迟
            # 上限 ≈1s，空轮询的 Redis 往返开销可忽略。
            task = queue.read(block_ms=min(self.read_block_ms, 1_000))
            if not task:
                continue

5）start 载体重投没有 resume 语义：每轮重投都经 start_flow 重新构造 context={} 的 state 并从头执行。state 构造（plaita/server/flow_worker.py:1322-1327）：

                engine_version=_engine_version(),
                tenant_id=current_tenant(),
                context={},
                status="running",
                start_time=datetime.now().isoformat(),
                invoker="worker"

随后无条件覆写盘上同一 execution_id 的行（plaita/server/flow_worker.py:1376），再以「无 saved_context」的方式启动（:1384）：

            self._persist_state_or_raise(execution_id, state, "start")

                result = execution.run_distributed(flow, params=params, execution_id=execution_id)

对照 resume 载体（plaita/server/flow_worker.py:1657-1662）——它才是「从盘上 checkpoint 续跑」：

            result = execution.run_distributed(
                flow,
                saved_context=state.context,
                resume_type=resume_type,
                resume_data=data,
            )

6）日志口径未随本次改动更新（plaita/server/flow_worker.py:663-667）：

        logger.warning(
            "执行 %s 节点执行失败，不终态化等待消息重投后重跑失败节点"
            "（第 %s/%s 次重试，重投间隔=claim_min_idle_ms）: %s",
            execution_id, attempt, self._node_retry_budget(), exc,
        )

7）用户文档同段仍写着 60s 回收节奏（docs-site/docs/distributed/flow-worker.md:77-82）：

- **载体**：at-least-once 消息重投本身。重试时执行**不终态化**（磁盘 state
  停在最后成功步 checkpoint——失败节点不写 context），run() **显式重投**同体新副本（先入队再 ack；
  `claim_min_idle_ms`（默认 60s）后被回收重投，resume 从 checkpoint 自然重跑
  失败节点。重投路径不依赖 pending 回收：消息可能已被竞争者的 `ExecutionLeaseError`
  分支 ack 释放，只留 pending 会导致重投永不发生（2026-10-09 plaita#52 实证）。
  重投失败（Redis 抖动）才退化为不 ack 留 pending；两种情况都不计 poison。

8）被本次删掉的旧节奏（git show b975109 -- plaita/server/flow_worker.py 的 - 行）：

-    ``run()`` 对本异常仿照 ``ExecutionLeaseError`` 处理：不 ack、不计
-    poison、留 pending，消息被回收（间隔=claim_min_idle_ms，默认 60s）后
-    resume 自然重跑失败节点。

9）既有测试不覆盖「重试节奏」，且 tests/unit/test_worker_node_retry.py:482 仍在断言旧语义（不 ack、留 pending）；显式重投的 4 个用例是 MagicMock 单点断言：

    def test_retryable_error_leaves_message_pending(self):

tests/unit/test_worker_node_retry.py:605-644 的 test_requeue_enqueues_then_acks / test_enqueue_happens_before_ack / test_enqueue_failure_falls_back_to_pending / test_ack_failure_still_reports_requeued 只断言 enqueue/ack 的调用顺序与退化分支，无一条测重试间隔或退避：

$ /tmp/plaita-audit-venv/bin/python -m pytest tests/unit/test_worker_node_retry.py -q
24 passed in 1.11s

10）复现 A（真实 run() 消费回路，resume 载体；节点失败用 mock 注入异常链）。命令与实测输出：

$ /tmp/plaita-audit-venv/bin/python /tmp/verify_retry_no_backoff.py
run_distributed 调用次数 : 5
相邻重试间隔(ms)        : [1, 1, 1, 1]
首次失败→终态化 耗时(ms): 60 (总观测 60 ms)
终态                   : error
error                  : {'message': '节点执行失败（重试 4 次后仍失败）: 执行节点a出错', 'node_retries': 4}
重试计数键残值          : 5
Traceback (most recent call last):
  File "/tmp/verify_retry_no_backoff.py", line 82, in <module>
    len(fake.xpending(q.stream_key, q.group_name).get("pending", []))
TypeError: object of type 'int' has no len()

（末尾的 traceback 是脚本最后一行打印 XLEN/PEL 自身的 bug——fakeredis 的 xpending 摘要此处返回 int——不影响上面六行；5 次预算 = 4 次重试，相邻间隔 ~1ms。）

11）复现 B（反证：节奏与 claim_min_idle_ms 无关，故日志/文档的「重投间隔=claim_min_idle_ms」是陈旧口径）。同一回路只改 claim_min_idle_ms：

$ /tmp/plaita-audit-venv/bin/python /tmp/verify_backoff_claim_irrelevance.py
claim_min_idle_ms=60000   尝试=5 相邻间隔(ms)=[2, 1, 1, 1] 总耗时=116ms 终态=error 计数=5
claim_min_idle_ms=1000    尝试=5 相邻间隔(ms)=[1, 1, 1, 1] 总耗时=59ms 终态=error 计数=5

12）复现 C（start 载体，真实 FlowExecution + 真实 code 节点，不 mock run_distributed；flow = start → n1(assignment 成功) → n2(code 每次抛 ConnectionError)）：预算照常耗尽，重放有界。

$ /tmp/plaita-audit-venv/bin/python /tmp/verify_start_retry_budget.py
code node registered: True
node_retry_budget = 5
------------------------------------------------------------------------
iter 0: 可重试失败 → 显式重投=True；计数键=1/5；盘上 status=running context={}
iter 1: 可重试失败 → 显式重投=True；计数键=2/5；盘上 status=running context={}
iter 2: 可重试失败 → 显式重投=True；计数键=3/5；盘上 status=running context={}
iter 3: 可重试失败 → 显式重投=True；计数键=4/5；盘上 status=running context={}
iter 4: 派发返回（终态化）status=error error=节点执行失败（重试 4 次后仍失败）: 执行节点n2出错了: ConnectionError: provider bli
------------------------------------------------------------------------
最终：status=error  error=节点执行失败（重试 4 次后仍失败）: 执行节点n2出错了: ConnectionError: provider blip
XLEN=1 PEL={'pending': 1, 'min': '1791530411060-0', 'max': '1791530411060-0', 'consumers': [{'name': 'worker-43326', 'pending': 1}]}

13）复现 D（start 载体每轮整条重放 + 盘上 context 被清空）：每轮真实执行的节点序列恒为 ['start','n1','n2']，start_flow 传给 run_distributed 的 saved_context 是本轮现场生成的值（不是盘上 checkpoint），盘上 context 恒为 {}。

$ /tmp/plaita-audit-venv/bin/python /tmp/verify_start_checkpoint.py
iter 0: 重投 | 本轮执行节点=['start', 'n1', 'n2'] | saved_context={'$INPUT': {}, '$BRANCH': None, '$NODE': {'start': None, 'n1': None}, '$PARENT': {}, 'EXPRESS_PREFIX': '$', '$FLOW_ID': 'f1', '$LAST_NODE': 'n1', '$GLOBAL': {'flow_id': 'f1'}, '$ENV': {}, '$EXECUTION_ID': 'exec-ckpt'} | 盘上 context={}
iter 1: 重投 | 本轮执行节点=['start', 'n1', 'n2'] | saved_context={'$INPUT': {}, '$BRANCH': None, '$NODE': {'start': None, 'n1': None}, '$PARENT': {}, 'EXPRESS_PREFIX': '$', '$FLOW_ID': 'f1', '$LAST_NODE': 'n1', '$GLOBAL': {'flow_id': 'f1'}, '$ENV': {}, '$EXECUTION_ID': 'exec-ckpt'} | 盘上 context={}
iter 2: 重投 | 本轮执行节点=['start', 'n1', 'n2'] | saved_context={'$INPUT': {}, '$BRANCH': None, '$NODE': {'start': None, 'n1': None}, '$PARENT': {}, 'EXPRESS_PREFIX': '$', '$FLOW_ID': 'f1', '$LAST_NODE': 'n1', '$GLOBAL': {'flow_id': 'f1'}, '$ENV': {}, '$EXECUTION_ID': 'exec-ckpt'} | 盘上 context={}
iter 3: 重投 | 本轮执行节点=['start', 'n1', 'n2'] | saved_context={'$INPUT': {}, '$BRANCH': None, '$NODE': {'start': None, 'n1': None}, '$PARENT': {}, 'EXPRESS_PREFIX': '$', '$FLOW_ID': 'f1', '$LAST_NODE': 'n1', '$GLOBAL': {'flow_id': 'f1'}, '$ENV': {}, '$EXECUTION_ID': 'exec-ckpt'} | 盘上 context={}
iter 4: 终态化 status=error | 本轮执行节点=['start', 'n1', 'n2'] | saved_context={'$INPUT': {}, '$BRANCH': None, '$NODE': {'start': None, 'n1': None}, '$PARENT': {}, 'EXPRESS_PREFIX': '$', '$FLOW_ID': 'f1', '$LAST_NODE': 'n1', '$GLOBAL': {'flow_id': 'f1'}, '$ENV': {}, '$EXECUTION_ID': 'exec-ckpt'}

影响

生产可达，且不需要任何误用：plaita worker 默认就是 --claim-min-idle-ms=60000 + --max-deliveries=5（第 3 条），节点级重试默认开启（回滚开关 PLAITA_DISABLE_NODE_RETRY 未设即生效，plaita/server/flow_worker.py:199-202、:650），判据只要求异常链里出现非超时的 NodeExecutionError（:352-381）——LLM/HTTP 节点遇到非超时类 5xx、连接重置即命中。

• 场景一（resume 载体，长流程）：执行已推进过若干步，后续节点撞上「持续数秒到数十秒的上游抖动」。5 次预算在 ~60ms 内烧完（第 10 条：调用 5 次、相邻间隔 [1, 1, 1, 1]），执行立刻落 error（error.node_retries=4）。改前同样是 4 次重试，但分散在 4×60s ≈ 4 分钟的回收窗口里，足以跨过这类抖动——现在跨不过（#50 记载该窗口在消息已被竞争者 ExecutionLeaseError 分支 ack 掉时不成立，那正是本次修复要解决的另一面；单 worker 与正常竞争下该窗口普遍存在）。可自愈的故障变成必失败，这是本次修复引入的可靠性回归。
• 场景二（start 载体，无 dedup_key 的 start 消息）：前序已成功的节点在预算内被整条重放最多 5 次（第 13 条：每轮节点序列恒为 ['start','n1','n2']）。前序节点若是 code/HTTP/写文件等有副作用的节点，副作用被重复执行 5 次，而两次之间只隔 ~1ms——上游或对端没有任何去重窗口。同一轮还会以 context={} 覆写盘上 running 行（:1324+:1376），排查时看到的 checkpoint 不是真实进度。
• 可观测口径失真：失败日志（:663-667）与 docs-site/distributed/flow-worker.md:77-82 都还写着「重投间隔=claim_min_idle_ms（默认 60s）」。运维按 60s 的节奏判断「还有几分钟才会耗尽预算」会得到完全错误的结论——实际上预算早已烧光、执行已是终态，或（start 载体）已经全量重放了 5 次。
• 不丢消息、不跨租户、有终态：本条不是 P0。代价集中在「可靠性回归 + 副作用重复 + 观测口径错误」，且该行为没有任何测试锁住（第 9 条 24 passed 全绿）。

不修的后果：任何持续数秒以上的上游抖动都会把一个本来能自愈的长流程直接打成 error，需要人工 resume_type=retry 唤醒；而带副作用前序节点的 start 流程会在无人察觉的情况下重复执行 5 次。

建议

修复方向在 plaita/server/flow_worker.py 的两个点，建议分两步：

• 退避（对症）：在 _requeue_retryable_failure（或 run() 的 except NodeExecutionRetryableError 分支）加入按 attempt 递增的退避。attempt 无需另算——_node_failure_retry_decision 已经把它放进异常对象（NodeExecutionRetryableError.attempt，:668-674）。最小修法是在入队前按 delay = min(cap, base * 2 ** (attempt - 1)) 延迟（base 取 1s 量级、cap 取 30–60s，与 claim_min_idle_ms 同量级）；更稳妥的做法是让队列支持延迟投递（入 ZSET 或其他延迟 Stream），避免在消费线程里真睡——单线程 worker 睡住会拖慢同队列其他任务，这一点需要在实现时明确取舍。
• 载体选择（治本的一半）：重投不应原样复制 body。type == "resume" 保持现状（从 checkpoint 续跑即可）；type == "start" 且盘上已存在同一 execution_id 的非空 checkpoint（或 running 行）时，应重投一份 resume 载体（{"type": "resume", ..., "resume_type": "continue"}），从而既避免从首个节点重放、也避免用 context={} 覆写 checkpoint；若要保留 start 语义，至少不能在已有 checkpoint 时先落 context={}。
• 顺带把日志（:663-667）与 docs-site/docs/distributed/flow-worker.md:77-82 的口径改成实际行为（重投间隔=退避值；start 载体的重放边界）。

验收条件：

• tests/unit/test_worker_node_retry.py 新增退避用例（沿用现有 fakeredis 夹具，用可注入的时钟/sleep 桩，禁止真 sleep 拉长测试）：模拟 resume 载体连续失败，断言相邻两次重投的间隔 ≥ 退避下限、且第 5 次失败后仍终态化 error、error.node_retries == 4（预算语义不变）。
• 新增 start 载体用例：预置盘上带 $LAST_NODE 的 running 行 + type="start" 消息，断言重投副本的 body["type"] == "resume"，或断言重投后 run_distributed 收到的 saved_context 等于盘上 checkpoint、且 on_node_start 序列不再包含已成功的前序节点。
• 复现脚本应变成：/tmp/verify_retry_no_backoff.py 的 相邻重试间隔(ms) 不再是 [1, 1, 1, 1]（应体现退避梯度）；/tmp/verify_start_checkpoint.py 从第二轮起 本轮执行节点 不再含 n1、盘上 context 不再被清成 {}；/tmp/verify_backoff_claim_irrelevance.py 两档 claim_min_idle_ms 之外，退避参数应能改变相邻间隔。

边界

• 本条不主张「预算永不耗尽 / 无界热循环」。实测 start 载体的计数键 1/5→2/5→3/5→4/5、第 5 轮终态化 error（第 12 条），预算正常耗尽，重放是有界的（最多 5 轮）。/tmp/repro_retry_budget.py 的 start 臂打印 counter=1 → counter=1 → …、终态=running，是因为该脚本把 run_distributed 换成了桩函数、其中每 3 次调用有 2 次返回「一步成功的 partial result」，于是 _process_execution_result 步进循环跑到 :1849-1851 的「节点成功推进即清零」语句，把计数反复清零；换成真实执行器（不 mock）后计数即单调递增并在第 5 轮终态化（第 12 条）。该脚本的 start 臂输出不能作为「计数器失效」的证据。
• 只修「显式重投缺退避 + start 载体缺 resume 语义」。不动队列层的 reclaim 语义（claim_min_idle_ms / XCLAIM / max_deliveries 死信阈值，plaita/server/task_queue.py），也不动 _handover_non_affine 的交接路径（草案 53 的地盘，两者只共享 enqueue + ack 的写法，根因与修复面不同）。
• 不动 _reset_node_retry_counter 的「节点成功推进即清零」组合语义（:1849-1851）及其在 G1 retry 唤醒处的调用（:1639）——那是「不同节点不共享预算」的刻意设计，本条只要求重投别把退避和载体语义一起丢掉。
• 本条针对的是基线区间 ea53e46..65f48f2 内的新增代码（b975109，2026-10-09 12:00），属该修复不完整而非存量缺陷：它把「不 ack 留 pending → 60s 后 reclaim」换成「立即入队 + ack」，却没有补上被删掉的节奏。start 载体的「整条重放」本身是存量行为——旧语义下 reclaim 到的同一条消息同样按 body.type 进入 start_flow（plaita/server/flow_worker.py:2601-2610），本次只是把间隔从 ~60s 压到 ~1ms，属放大而非新增。
• 与既有 issue 的关系（gh issue list -R jeffkit/plaita --state all 现有 34 条）：#52（OPEN，执行终态滞后）是本次 b975109 声称要修的原缺陷，本条是该修复的副作用（节奏被删），不是同一根因，相关而非重复；#50（OPEN）讲的是修复前重投载体被 ExecutionLeaseError 的 ack 吃掉；#48（OPEN）是 reclaim 触发重投风暴，方向相反；#47（OPEN）是 reclaim 不查活跃租约；#49（OPEN）是抢单缺本机盘预检后的退避升档（不同函数）；#22/#23/#24/#30 已 CLOSED 且主题无关。建议本条与 #52 互相引用即可，不必并入。
• 与本人既有草案的区别：草案 33/34/35 是 plaita/dsl/codeflow/_emit.py 的 IR→源码发射保真问题（位点在 DSL 发射器），草案 51 是死信守卫的租户命名空间、53 是非亲和交接无终止条件——本条位点在 plaita/server/flow_worker.py 的显式重投与 start 入口，代码行与后果均不重叠。