fix(service): #23 新增的调度 dedup_key 以 fire_schedule 墙钟秒为基准而非到期时点 next_run_at——同一个 due 的重触发拿到不同键、两次均 claimed，同一次 cron 到期双跑；manual 与 cron 同秒互相吞并

Start in plaita/server/services/schedule_service.py at fire_schedule, where dedup_key is built from the wall-clock second rather than next_run_at, and at _fire_if_due, which already computes the due time. Then read tests/unit/test_issue23_start_lease_window.py, whose existing dedup-key tests assert the buggy behaviour. Done means two fires of the same due produce the same dedup_key and only one execution row, with cross-period and manual-vs-cron cases covered. The manual-key semantics need a maintainer decision before coding.

优先级 P2 · 依赖：无

基线 HEAD = 65f48f2

Summary

fire_schedule 新增的调度幂等键（3ce6e6f，属 #23 修复）取的是调用时刻的墙钟秒，
而不是本次触发的到期时点 next_run_at。于是「同一个 next_run_at 被触发两次」时，
两次触发落在不同秒 → 两个不同的 dedup_key → worker 两次都 SET NX 成功、各自认领、
各自铸 execution_id，同一次 cron 到期把整条 flow 双跑。该键只在「同一条消息被
XCLAIM 重派/显式重投」这一种情形下收敛。

1）键 = schedule_id + 调用墙钟秒（plaita/server/services/schedule_service.py:109-111）：

    message["dedup_key"] = "sched:{schedule_id}:{fired_at}".format(
        schedule_id=schedule_id, fired_at=now_ms // 1000
    )

2）fired_at 的来源就是 fire_schedule 被调用那一刻的 datetime.now()，与本次到期时点
无关（plaita/server/services/schedule_service.py:83-84）：

    now = datetime.now()
    now_ms = int(now.timestamp() * 1000)

3）新增注释自己声明了这个键的适用范围，也把「下一次到期时间戳必然不同」当成免误吞的
保证——但反过来同一个到期被触发两次时，时间戳同样不同（plaita/server/services/schedule_service.py:102-108）：

    # 确定性幂等键（#23）：start 消息此前无 execution_id 也无 dedup_key，
    # 消息被 XCLAIM 重派后 worker 就地铸造新 id——同一逻辑 run 出现两个平行
    # 执行、整条 flow 从头双跑。键 = schedule_id + 触发时点（epoch 秒）：
    # 同一次到期的重投/重派消息收敛到同一 execution；下一次 cron 到期时间戳
    # 必然不同，不会误吞后续周期（worker 侧 SET NX EX 7d 认领，见
    # FlowWorker._claim_start_dedup）。时间戳取秒级：cron 语义最小粒度是分钟，
    # 秒级已足够区分两次到期，且对时钟小幅回拨容忍。

4）入队（plaita/server/services/schedule_service.py:113-115）排在 next_run_at 回写
（:141-150）之前，且回写是 best-effort、异常只 warning、不重试也不推进
next_run_at：

    try:
        # 统一走 enqueue_task：消息信封带 schema_version，老格式（无该字段）按 v1 处理
        msg_id = enqueue_task(redis_client, queue_name, message)

    # 回写触发状态；错过的周期不补偿——next_run_at 直接跳到现在之后
    try:
        updates = {
            "last_fired_at": now.isoformat(),
            "last_enqueue_ok": "1" if enqueue_ok else "0",
        }
        if trigger_kind == "cron":
            updates["next_run_at"] = str(next_run_after(schedule["cron"], now))
        redis_client.hset(SCHEDULES_KEY, key=schedule_id, value=json.dumps(_merge(schedule, updates), ensure_ascii=False))
    except Exception:
        logger.warning("调度 %s 状态回写失败", schedule_id, exc_info=True)

进程在 enqueue_task 与 hset 之间被 SIGTERM / 崩溃 / Redis 写瞬断，next_run_at 就停在
旧值，而消息已经躺在队列里。

5）到期判定读的正是那个没被推进的旧值，拦不住第二次
（plaita/server/services/schedule_service.py:295-296）：

        if now_ms < next_run_at:
            return

6）唯一的互斥是 30 秒锁，过期即放行重触发（plaita/server/services/schedule_service.py:298-302），
紧接着就是 cron 侧唯一的触发点（:304）：

        # 互斥：多实例部署时保证同一时点只触发一次
        lock_key = SCHEDULE_LOCK_KEY.format(schedule_id=schedule_id)
        lock_holder = self._service_info.instance_id if self._service_info else "schedule"
        if not self._redis_client.set(lock_key, lock_holder, nx=True, px=30_000):
            return

        fire_schedule(self._redis_client, schedule, self.queue_name, trigger_kind="cron")

SCHEDULE_LOCK_KEY = "plaita:schedule:lock:{schedule_id}"（schedule_service.py:39）。

7）worker 侧对两个不同的键各自认领成功，没有任何跨触发的同源性校验
（plaita/server/flow_worker.py:1086-1097；START_DEDUP_TTL_SECONDS = 7 * 86400，
flow_worker.py:1063）：

        if not dedup_key or redis_client is None or not hasattr(redis_client, "set"):
            return ("inactive", None)
        key = self._start_dedup_key(dedup_key)
        try:
            claimed = redis_client.set(
                key, execution_id, ex=self.START_DEDUP_TTL_SECONDS, nx=True
            )
        except Exception as e:  # noqa: BLE001 — 认领瞬断：不阻塞启动（退化为存量行为）
            logger.warning("start 幂等键认领失败（按未启用处理）: %s: %s", key, e)
            return ("inactive", None)
        if claimed:
            return ("claimed", None)

8）fire_schedule 的全仓生产调用点只有两处，cron 与 console「立即触发」共用键空间，
而 manual 路径不持步骤 6 的锁（plaita-console/backend/api/schedules.py:332）：

    msg_id = fire_schedule(redis, schedule, TASK_QUEUE_NAME, trigger_kind="manual")

反证命令：

$ grep -rn "fire_schedule" --include='*.py' . | grep -v "/tests/\|test_"
./plaita/server/services/schedule_service.py:68:def fire_schedule(
./plaita/server/services/schedule_service.py:304:        fire_schedule(self._redis_client, schedule, self.queue_name, trigger_kind="cron")
./plaita-console/backend/api/schedules.py:36:        fire_schedule,
./plaita-console/backend/api/schedules.py:48:            fire_schedule,
./plaita-console/backend/api/schedules.py:332:    msg_id = fire_schedule(redis, schedule, TASK_QUEUE_NAME, trigger_kind="manual")

（:36 / :48 是 import，:68 是定义；真正的触发调用只有 :304 与 :332。）

9）复现一：真实 _tick → _fire_if_due → fire_schedule 入口臂。
命令：

cd /Users/jeff.jie/projects/infra4agent/plaita && /tmp/plaita-audit-venv/bin/python /tmp/verify_f3_tick.py

脚本预置 HASH 中 next_run_at 为 300s 前的值、enabled=True，让首次状态回写抛异常
（模拟 SIGTERM 落在 enqueue 与 hset 之间），等 30s 互斥锁过期后再 tick 一次。
实跑输出（逐字）：

初始 next_run_at=1791530071848（已过期 300s，enabled=True）
锁键初始存在? 0
tick1 后 next_run_at=1791530071848
tick1 触发记录数=1 keys=['sched:s1:1791530371']
锁 TTL 剩余=29999ms
等待互斥锁过期（真实场景为 30s；此处 pttl 归零后即可）...
tick2 后队列消息数 1 → 2
tick2 触发记录数=2 keys=['sched:s1:1791530402', 'sched:s1:1791530371']
   msg=1791530371851-0 dedup_key=sched:s1:1791530371
   msg=1791530402970-0 dedup_key=sched:s1:1791530402
两次 due 相同（1791530071848）、dedup_key 相同? False
独立执行行数=2 statuses=['completed', 'completed']

即：两次触发的 next_run_at 完全相同（1791530071848，都是同一个过期 due），
但拿到 sched:s1:1791530371 与 sched:s1:1791530402 两个键，worker 两次都放行，
盘上落下 2 条 completed 执行行。（键中的秒级数字取自墙钟，每次实跑会不同，形态不变。）

10）复现二：三臂对照。命令：

cd /Users/jeff.jie/projects/infra4agent/plaita && /tmp/plaita-audit-venv/bin/python /tmp/verify_f3_sched_dedup.py

实跑输出（逐字）：

场景 A：同一条消息原样重投（同 body → 同 dedup_key）
    dedup_key=sched:s1:1791530404
    msg=1791530404502-0 key=sched:s1:1791530404 delivery=1 → run_distributed=1 exc=None rows={'2536d3c11fea4084b3e914be19f55183': 'completed'}
    msg=1791530404503-0 key=sched:s1:1791530404 delivery=1 → run_distributed=0 exc=None rows={'2536d3c11fea4084b3e914be19f55183': 'completed'}

场景 B：同一次到期触发两次（next_run_at 回写丢失 + 30s 锁过期）
    回写失败后 HASH 里存在 s1? False
    队列消息数=2
      msg=1791530404553-0 dedup_key=sched:s1:1791530404
      msg=1791530405657-0 dedup_key=sched:s1:1791530405
    两键相同? False
    msg=1791530404553-0 key=sched:s1:1791530404 delivery=1 → run_distributed=1 exc=None rows={'701ac7f6960c4c9da177db80261ed83d': 'completed'}
    msg=1791530405657-0 key=sched:s1:1791530405 delivery=1 → run_distributed=1 exc=None rows={'701ac7f6960c4c9da177db80261ed83d': 'completed', '4aff73b9be324d50a3fb0d4a83f8791e': 'completed'}

场景 C：同一秒内 cron + manual（console「立即触发」）
    msg=1791530405670-0 kind=cron dedup_key=sched:s1:1791530405
    msg=1791530405671-0 kind=manual dedup_key=sched:s1:1791530405
    两键相同? True
    msg=1791530405670-0 key=sched:s1:1791530405 delivery=1 → run_distributed=1 exc=None rows={'d6b0090a74654fbb8a9c8c2ac02bfc5f': 'completed'}
    msg=1791530405671-0 key=sched:s1:1791530405 delivery=1 → run_distributed=0 exc=None rows={'d6b0090a74654fbb8a9c8c2ac02bfc5f': 'completed'}

场景 A 证明该键并非全然无效——同 body 重投确实收敛；场景 B 证明它对「同一次到期被重
触发」无效（两键不同、两次都 run_distributed=1、盘上两行）；场景 C 证明同一秒内 cron 与
manual 会共用同一个键，后到者被静默吞掉（run_distributed=0），而 API 仍返回
success + msg_id。

11）文档中随本次一并写入的承诺，在重触发面上不成立
（docs-site/docs/distributed/flow-worker.md:139-142）：

- **调度服务是唯一的自动加键方**（#23）：`fire_schedule`（cron 循环与 console
  「立即触发」共用）以确定性键 `sched:{schedule_id}:{触发时点秒}` 入队——
  同一次到期的消息重派/重投收敛到同一 execution，不产生平行执行；下一次
  cron 到期时间戳不同，不会被误吞。手动 `POST /executions` 仍不自动加键。

12）仓库自带测试没有覆盖「同一个 due 被触发两次」这一判据
（tests/unit/test_issue23_start_lease_window.py:362-396）：只断言键前缀、后缀是数字，
以及 time.sleep(1.1) 后两次触发的键不同（即恰好把缺陷行为当成了期望行为）：

        assert msg["dedup_key"].startswith("sched:s9:")
        # 键值 = schedule_id + 触发时点（秒级 epoch）：确定性、跨周期不同
        suffix = msg["dedup_key"].rsplit(":", 1)[-1]
        assert suffix.isdigit()

    def test_dedup_key_follows_fired_at_not_wall_clock_drift(self):
        """两次触发时间不同 → 键不同（下一周期不被误吞）。"""
        fake = fakeredis.FakeRedis(decode_responses=True)
        fire_schedule(fake, self._schedule(), QUEUE, trigger_kind="cron")
        time.sleep(1.1)
        fire_schedule(fake, self._schedule(), QUEUE, trigger_kind="cron")
        entries = fake.xrange(QUEUE)
        keys = {
            json.loads(
                f["payload"] if isinstance(f["payload"], str) else f["payload"].decode()
            )["dedup_key"]
            for _, f in entries
        }
        assert len(keys) == 2

全文件无 next_run_at / SCHEDULE_LOCK_KEY 相关断言（反证命令）：

$ grep -n "next_run_at\|lock_key\|SCHEDULE_LOCK" tests/unit/test_issue23_start_lease_window.py
（零命中）

影响

生产可达，集群档默认链路，无 feature flag。 cron 侧 ScheduleService._run_loop →
_tick（schedule_service.py:262）→ _fire_if_due（:273）→ fire_schedule（:304）；
manual 侧 console /schedules/{id}/trigger（集群档，schedules.py:332）。

触发条件不是「时钟精度」而是两步非原子：enqueue_task（:115）成功、而
hset（:148）没跑成。命中方式都很日常——Redis 写瞬断（主从切换、连接池超时、实例重启）、
或 worker/调度服务恰在两步之间收到 SIGTERM、容器被 OOM kill、滚动发布重启。
30 秒的互斥锁让这段窗口足够长：只要重触发发生在锁过期之后，第二次一定拿到新键。

代价：

• 同一次 cron 到期把整条 flow 从头双跑两遍。这正是 #23 想消灭的双跑，只是残留在了
  「回写失败」这个失败面上；对下游是重复的副作用（重复发信、重复写库、重复调外部 API），
  而 next_run_at 已停在旧值，锁一过期就可能被反复重触发。
• 两条平行执行在盘上都是独立的 status=completed 行，事后无法从状态区分哪条是多余的，
  只能靠人工对账。
• 反向的过度去重同样存在：cron 到期那一秒如果恰好有人点了 console「立即触发」，
  两条消息共用同一个键，manual 那条被静默吞掉，run_distributed=0，而 API 仍返回
  {"success": true, "msg_id": ...}——操作者以为触发了，实际没跑，且没有任何告警。
• 唯一留下的痕迹是触发历史里两条 fired_at 不同的 fire 记录与两个不同 dedup_key，
  以及 last_enqueue_ok（回写失败时它本身也不会被写）。

不修的后果：调度侧的「确定性幂等键」只能在消息被重派时生效，而真正需要它兜住的那类
双跑（同一次到期被触发两次）恰好不被覆盖；同时它还引入了一个新的静默吞单面。

建议

修复方向（plaita/server/services/schedule_service.py，一处）：

• 把键的时间基准从「调用墙钟」换成「本次到期时点」。fire_schedule 已经能从
  schedule 里读到 next_run_at；让调用方把到期时点显式传进来（cron 侧在 _fire_if_due
  里已算出 next_run_at，:280-293，直接传参即可），键写成
  sched:{schedule_id}:{due} 或 sched:{schedule_id}:{due}:{trigger_kind}。这样同一个
  due 无论被触发几次都得到同一个键，worker 第二次会 hit 而不二次 start；
  下一次 cron 到期的 due 不同，不会误吞。
• manual 需要与 cron 分开：trigger_kind="manual" 是用户显式要求「立刻再跑一次」，
  把它和 cron 的到期混进同一键空间是场景 C 的成因。建议 manual 不复用 due（例如带
  manual:{time_ns} 或干脆不自动加键，交回调用方决定），或在键里带上 trigger_kind。
  这一点需与维护者确认语义后再定；若沿用「manual 也自动加键」，至少不能与 cron 同秒碰撞。
• 顺手收紧回写失败面（可选，见「边界」）：入队与回写之间没有任何补偿，
  可在 hset 失败时把 last_enqueue_ok=0 记入触发历史（当前历史写入在 :131-138，
  与回写是同一次尝试），让运维至少能看见「下一次 tick 会重触发」。本单不强制要求。
• 不改 worker 侧 _claim_start_dedup / _start_dedup_key 的语义与 7 天 TTL，不改
  SCHEDULE_LOCK_KEY 的 30s 窗口，不动 next_run_after 的排程语义。

验收条件：

在 tests/unit/test_issue23_start_lease_window.py 新增一条用例（沿用文件里已有的
fakeredis + _schedule() 夹具），覆盖「同一个 due 触发两次」：对同一份 schedule
（next_run_at 相同）连续调用两次 fire_schedule，断言：

• 队列里两条消息的 dedup_key 相同（当前实现为不同，是红）；
• 端到端（沿用 test_keyed_message_converges_on_requeue 的 MemoryExecutionStorage 回路）
  消费两条后，盘上只有 1 行执行、且第二个 dedup_key 命中 ("hit", ...)、
  run_distributed 只被调用 1 次。

同时补一条反向用例，锁住「跨周期不被误吞」：两个不同 next_run_at（或不同 cron 到期）
触发时，dedup_key 必须不同，两条消息都要真正跑起来。

再补一条 manual 语义用例：同一秒内先 cron 后 manual（或先 manual 后 cron），断言两者
不互相吞并（各自产生一次执行，或在明确约定 manual 不复用调度键时断言 manual 走独立键）。

/tmp/verify_f3_tick.py 在上述修复后应当输出：两次 tick 的 dedup_key 相同、
独立执行行数=1；/tmp/verify_f3_sched_dedup.py 场景 B 应当变为
两键相同? True + 第二次 run_distributed=0，场景 C 应当变为 两键相同? False。

边界

• 只修 plaita/server/services/schedule_service.py 里 fire_schedule 的 dedup_key
  取值基准（以及 manual 与 cron 的键空间划分）；不动 worker 侧认领逻辑、不动
  next_run_at 的排程/自愈语义（:280-293）、不改调度互斥锁与 TTL、不引入新的重试或
  补偿子系统。
• 与 #23（CLOSED）的关系：#23 讲的是「start 消息无租约无占用标记 → 多 worker 池下

  60s 的 start 被空闲 worker XCLAIM 重派，整条 flow 双跑」，修法是租约 + worker 侧
  _claim_start_dedup；本单是那次修复里新增的调度侧加键取错了时间基准（3ce6e6f
  引入，属 ea53e46..65f48f2 区间），是 #23 修复在失败面上的不完整，不是同一个缺陷。
  #23 的结论性内容无需修改，若要补信息应作为其补充评论；但本单的修复位点、验收判据与之
  完全无重叠，故按新单提。

• 与 #46（OPEN）不同根：#46 是扩展服务启动器缺 logging 配置（delay/schedule/approval/
  callback/queue 全域静默），属可观测性接线；本单是键的时间基准错误。二者都涉及
  schedule_service，但一个改日志初始化、一个改幂等键，无代码重叠。
• 与 #47 / #50 / #52（均 OPEN）不同根：这三单都在 worker 侧消费链上（#47 reclaim 不查
  活跃租约、#50 重投载体被租约冲突 ack 吃掉、#52 执行终态滞后）。本单的修复在生产侧
  fire_schedule，不动 worker 任何一行；即使 worker 侧三单全部修好，同一个 due 的双触发
  仍会产生两个不同键并各自被正常认领，双跑照旧。
• 与 #22 / #24 / #29 / #30（均 CLOSED）无关：#22/#29/#30 是 nodes 安全面（沙箱白名单
  未接线、js 语言绕开沙箱、子进程 env 继承），#24 是租户凭据路由；与调度加键无交集。
• 与本批 _emit.py 发射器族（33 / 34 / 35）不同根：那三单是 emit_source 的 IR→源码
  保真问题（条件逆映射、事件字段别名、end 节点 resultType/error 丢失），属 codeflow
  反编译；本单是运行时调度幂等键。文件、组件、链路均不同。
• 与 14-schedules-tenant-cross-reach.md 不同根：那一单是 /api/schedules 单条端点不按
  租户过滤取数（越权读/触发他租户调度），同一文件但属鉴权面；本单不涉及租户过滤，修复也
  不触碰该端点。
• 生产发生率未量化：本机无真实 Redis 与生产调度部署，只验证了机制与真实入口链路
  （_tick → _fire_if_due → fire_schedule、以及 console manual 调用点），没有统计
  last_enqueue_ok=0 / 回写失败的实际频次——这是本单保留 P2 而非 P1 的原因。
  一旦确认生产已出现该双跑，应升为 P1。
• 复现基于 fakeredis 与会抛异常的回写桩来等价模拟 Redis 写失败 / 进程在两步之间被杀；
  不承诺对「hset 已写入但客户端未收到」这类模糊失败的分析，那需要单独的幂等设计。
• 查重：gh issue list -R jeffkit/plaita --state all --limit 100 中与 fire_schedule /
  dedup_key / next_run_at 相关的既有关闭单只有 #23，#23 正文未涉及键的时间基准，不构成
  重复；本批其余 64 份草案中命中 dedup_key|next_run_at|fire_schedule 的仅
  14-schedules-tenant-cross-reach.md（租户取数）、38-subscription-ttl-keepttl-loss.md、
  54-event-dedup-loses-resume.md、41-redis-bus-task-ref.md（均为 event_filter / 订阅
  去重族），无同根条目。