《AI Agent 工程实战》5.2 事务与并发：run 状态机原子更新、行锁防重复执行

5.2 事务与并发：run 状态机原子更新、行锁防重复执行

上一节把 runs 表建出来了，但它现在只是几行躺着的状态字段。真实系统里，一张 runs 表同时被 API 进程（创建、取消）、多个 worker 进程（抢占、执行、回写）读写。只要 worker 多于一个，「谁该跑这次运行」就成了一个并发问题——而 Agent 运行一次可能花掉几十秒和真金白银的 token，重复执行不是「多花点电」，是重复扣费、重复发通知、重复写副作用。

本节把 ResearchCopilot 推进到并发正确性：用「条件 UPDATE + RETURNING 的返回行数」把 run 的抢占做成一次原子跃迁，真跑两个并发 worker 抢同一个 run，验证只有一个抢到、另一个空手而归。

5.2.1 run 是一台状态机

先把 runs.status 的取值范围钉死。它是一个有限状态机，每一跳都有明确的前置条件：

关键约束是：跃迁只能从「当前状态」出发。一个 succeeded 的 run 不能再变成 running，一个已经被抢走的 queued 不能再被第二个 worker 抢走。把这条约束写成代码，就是本节的全部内容——而它恰好能由数据库用一条语句、原子地完成。

状态机还有一个容易被低估的价值：它让「系统现在在干什么」变成一次查询就能回答的问题。有了 status 这个受约束的枚举（CHECK 保证只有五个合法值），运维就能问「有多少 run 卡在 running 超过五分钟」「failed 的占比是不是在涨」；而如果状态是一堆散落的布尔字段（is_running、is_done、has_error），这些问题就退化成对着一堆互相矛盾的标志位猜谜。把状态收敛成一个枚举列，是让系统可观测的第一步。

5.2.2 抢占执行权：一条 UPDATE 就是一次 CAS

worker 要做的事是「如果这个 run 还是 queued，就把它变成 running 并归我」。用 SQL 写出来只有一句：

CLAIM = ("UPDATE runs SET status='running', started_at=now(), attempts=attempts+1 "
         "WHERE id = %s AND status = 'queued' RETURNING id")

def claim(conn, run_id) -> bool:
    return conn.execute(CLAIM, (run_id,)).fetchone() is not None

这一句同时做了三件事，缺一不可：

1. WHERE status = 'queued' 是前置条件，等价于比较并交换（compare-and-swap）里的「期望值」。只有状态还是 queued 时，这一行才满足更新条件。
2. RETURNING id 让「有没有更新到」可判定。UPDATE 返回的 cursor，fetchone() 拿到行就说明这一行是我改的；拿到 None（即影响 0 行）说明别人抢先了或状态已变。这就是「谁抢到」的判据——不需要再查一次。
3. 整条语句在行锁下原子执行。PostgreSQL 执行 UPDATE 时会给命中的行加排他锁；并发的第二个 UPDATE 会阻塞，直到第一个提交，然后在锁下重新判断 WHERE 条件（READ COMMITTED 的语义）。

第 3 点是最容易被忽略、也最关键的一点。它意味着：两个 worker 同时发这条 UPDATE，数据库不会让它们都成功——第二个会等第一个提交，重新评估后发现 status 已经是 running，条件不再满足，于是影响 0 行。抢占的互斥性由数据库的行锁保证，而不是由应用层的「检查—动作」保证。

这就是「乐观并发控制」在数据库里最朴素的形态：不加显式的锁语句，只把「期望状态」写进 WHERE，让数据库在更新那一刻替你做判定。它比应用层加锁简单得多，也不会像「先锁再查」那样引入死锁面。

5.2.3 为什么不能先 SELECT 再 UPDATE

初学者最自然的写法是分两步：先 SELECT status，看到是 queued，再 UPDATE。这个写法在单进程测试里永远正确，一上并发就崩。原因是一个经典的检查—动作竞态（check-then-act race）：

worker A: SELECT status -> 'queued'
worker B: SELECT status -> 'queued'      # B 在 A 更新前也读到了旧值
worker A: UPDATE ... SET running          # A 先写
worker B: UPDATE ... SET running          # B 也写，同一个 run 被跑了两遍

问题出在：SELECT 和 UPDATE 之间没有任何锁，两个事务可以都读到 queued，然后都去写。真跑这个反例——用 threading.Barrier 强制两个 worker 都先读完状态再动手：

import threading, psycopg
DSN = "postgresql://postgres:postgres@localhost:15435/research"

log, bar = [], threading.Barrier(2)
def naive(worker):
    with psycopg.connect(DSN) as conn:
        seen = conn.execute("SELECT status FROM runs WHERE id=%s", (rid,)).fetchone()[0]
        bar.wait()                       # 两个 worker 都读到状态后才继续
        if seen == "queued":
            conn.execute("UPDATE runs SET status='running' WHERE id=%s", (rid,))
            conn.commit()
            log.append(f"{worker} 看到 '{seen}' -> 开始执行")
        else:
            log.append(f"{worker} 看到 '{seen}' -> 跳过")

ts = [threading.Thread(target=naive, args=(w,)) for w in ("A", "B")]
[t.start() for t in ts]; [t.join() for t in ts]
for line in sorted(log):
    print(line)

真跑输出（PostgreSQL 17.11）：

A 看到 'queued' -> 开始执行
B 看到 'queued' -> 开始执行
最终状态: running (但被跑了两次)

两个 worker 都看到 queued、都执行了。最终状态看起来「正确」（确实是 running），但副作用已经发生了两次——这正是这类 bug 最阴险的地方：状态字段永远显示正常，只有账单和用户投诉会告诉你出过事。

一个常见的「补丁」是把 UPDATE 的 WHERE 改成 WHERE id=%s AND status='queued'，然后看影响行数再决定要不要执行。这其实已经等价于 5.2.2 的原子写法了——只要把「判断」和「写入」合并进同一条语句，竞态就消失了。判据是「我改到了这一行」，而不是「我读到过这个状态」。

再补一句：即使把两步塞进同一个事务，只要隔离级别是 READ COMMITTED，SELECT 读到的仍可能是「别人已经看到、但还没提交」的旧快照，两步之间依旧有窗口。想靠事务「包起来」解决竞态，是另一个常见的错觉——要么用一条语句，要么用 SELECT ... FOR UPDATE 显式加锁，没有第三条路。

5.2.4 真跑：两个 worker 抢同一个 run

把原子抢占放进两个真实线程，让 worker A 抢到后故意不提交（模拟它正在执行），worker B 在 A 持锁期间发起同样的抢占：

import threading, time, psycopg
DSN = "postgresql://postgres:postgres@localhost:15435/research"
CLAIM = ("UPDATE runs SET status='running', started_at=now(), attempts=attempts+1 "
         "WHERE id=%s AND status='queued' RETURNING id, attempts")

res, a_holding = {}, threading.Event()

def worker_a():
    with psycopg.connect(DSN) as conn, conn.transaction():
        res["A"] = conn.execute(CLAIM, (rid,)).fetchone()
        a_holding.set()
        time.sleep(1.2)          # 拿着行锁不放，模拟 A 正在执行

def worker_b():
    a_holding.wait()             # 等 A 抢到后再发起，保证并发窗口
    t0 = time.perf_counter()
    with psycopg.connect(DSN) as conn, conn.transaction():
        res["B"] = conn.execute(CLAIM, (rid,)).fetchone()
    res["B_ms"] = round((time.perf_counter() - t0) * 1000, 1)

ta, tb = threading.Thread(target=worker_a), threading.Thread(target=worker_b)
ta.start(); tb.start(); ta.join(); tb.join()

真跑输出：

run_id = a0bbc912-7fb3-49d1-80f8-f24c3fd52f66
worker A 抢到: (UUID('a0bbc912-7fb3-49d1-80f8-f24c3fd52f66'), 1)
worker B 抢到: None  (阻塞 1222.5 ms 后空手而归)
最终状态: ('running', 1)

三个数字把机制讲透了：

• worker A 拿到了一行（(UUID(...), 1)，即 id 与 attempts），说明抢占成功。
• worker B 拿到 None，即影响 0 行——它没抢到。
• B 阻塞了 1222.5 ms，几乎正好是 A 持锁的 1.2 秒。这不是巧合：B 的 UPDATE 在行锁上等 A 提交，A 一提交，B 在锁下重新评估 WHERE status='queued'，发现已经是 running，条件不满足，返回 0 行。

最后状态是 running、attempts=1——恰好被执行一次。这正是我们要的语义。顺带注意 attempts 的自增写在同一条语句里（attempts = attempts + 1），所以它也是原子的，不会因为并发而丢计数；如果换成「Python 里先读 attempts、加一、再写回」，同样的竞态会立刻回来。

5.2.5 从队列取下一个任务：FOR UPDATE SKIP LOCKED

5.2.4 抢的是「一个已知 id 的 run」。但 worker 平时面对的是「队列里有一堆 queued，我该取哪个」。天真的做法是先 SELECT ... WHERE status='queued' LIMIT 1，再 UPDATE 抢占——这又回到了「检查—动作」两步，两个 worker 会取到同一条。想用行锁互斥，直觉是 SELECT ... FOR UPDATE，但它会让第二个 worker 阻塞等第一个，队列一长，worker 就被串行化了。

正解是 FOR UPDATE SKIP LOCKED：取的时候加锁，但遇到已被别人锁住的行直接跳过，去找下一条。多个 worker 于是各取一条、互不阻塞。

PULL = ("SELECT id, attempts FROM runs WHERE workspace_id=%s AND status='queued' "
        "ORDER BY created_at LIMIT 1 FOR UPDATE SKIP LOCKED")

真跑：先塞 10 条 queued，两个 worker 用 threading.Barrier 同时取：

queued runs: 10
worker A 取到: a403a910-4caf-4064-9fe2-6b1d88aeb300
worker B 取到: 9983a529-fa55-4180-aa9a-f5540bf214f7
两者不同: True

两个 worker 取到了不同的 run。注意 FOR UPDATE 加的锁保持到事务结束——所以「取任务」和「改状态为 running」应该在同一个事务里完成：取到后立刻 UPDATE ... SET status='running'，再提交。这样「取」与「占」是一体的，不存在「取了但没占」的窗口。三种取任务写法的取舍：

SKIP LOCKED 的语义边界要记牢：它跳过的是「已经被别的事务锁住」的行，不是「状态不是 queued」的行——状态过滤仍然靠 WHERE status='queued'。两者配合，才构成一个不重不漏的取任务循环。

5.2.6 完整的状态跃迁与终止

抢占只是状态机的一跳。把「正常结束」「失败」「取消」也写成同样的条件 UPDATE，整台机器就完整了：

CLAIM   = ("UPDATE runs SET status='running', started_at=now(), attempts=attempts+1 "
           "WHERE id=%s AND status='queued' RETURNING id")
SUCCEED = ("UPDATE runs SET status='succeeded', finished_at=now() "
           "WHERE id=%s AND status='running' RETURNING id")
FAIL    = ("UPDATE runs SET status='failed', error=%s, finished_at=now() "
           "WHERE id=%s AND status='running' RETURNING id")
CANCEL  = ("UPDATE runs SET status='cancelled', finished_at=now() "
           "WHERE id=%s AND status IN ('queued','running') RETURNING id")

真跑一串跃迁，看每一步的成败：

初始: ('queued', 0)
claim   -> True
重复 claim -> False (第二个 worker 抢不到)
succeed -> True
重复 succeed -> False (终态不可再跃迁)
cancel  -> False (已终态，取消无效)
最终: ('succeeded', 1)

每一步的返回值都是一次合法性判定：重复 claim 返回 False，是因为状态已是 running；对终态再 succeed 或 cancel 返回 False，是因为 WHERE 里的状态条件不满足。状态机的守卫条件直接编码在 SQL 的 WHERE 里，不需要在 Python 里写一堆 if，也就不存在「Python 判断和数据库状态不一致」的窗口。

还有两个工程细节值得记住：

• 取消 running 的 run 不能只改数据库。CANCEL 把状态改成 cancelled 只是「发出信号」，真正在跑的引擎还得读到这个信号并主动中止（卷一的 cancel 路径）。数据库负责「只能取消一次」，引擎负责「取消得掉」。这块的取消与超时留到 8.3 取消、超时与泄漏防护 。
• 重试不是把 failed 改回 queued 了事。attempts 是防无限重试的闸门，真正的重试策略（退避、死信）见 7.2 重试与死信队列 。本节只保证状态跃迁本身是原子的。

5.2.7 真实环境与验收清单

本节所有输出都来自本机真连的 PostgreSQL 17.11（容器 ab-pg，连接串 postgresql://postgres:postgres@localhost:15435/research）。几个真实环境的补充：

• READ COMMITTED 是默认隔离级别，本节的阻塞后重评估语义正是它提供的。不要为了「更安全」把隔离级别提到 SERIALIZABLE——那会让并发抢占频繁抛序列化失败，得不偿失；单条条件 UPDATE 在 READ COMMITTED 下已经足够。
• 长事务是敌人。worker A 持锁 1.2 秒在演示里没问题，但真实执行一次 Agent 运行要几十秒，绝不能在事务里跑完整段 Agent 逻辑——那会把行锁攥住几十秒，拖垮所有并发 worker。正确做法是：用一次短事务完成「抢占」（把状态改成 running 后立刻提交），执行阶段在事务之外，最后再用一次短事务回写终态。
• 抢占要配超时兜底。如果 worker 抢到 running 后进程崩了，run 会永远卡在 running。生产里需要一个「僵尸回收」任务，把 running 超过 N 分钟未更新的 run 重置为 queued——这属于 7.3 定时任务 的范畴。

本节验收清单：

• 抢占由单条 UPDATE ... WHERE status='queued' RETURNING id 完成，判据是返回行数
• 两个并发 worker 抢同一个 run，恰好一个成功、一个影响 0 行
• 「先 SELECT 再 UPDATE」的反例能复现出「跑两遍」
• 队列取任务用 FOR UPDATE SKIP LOCKED，两个 worker 取到不同 run
• 终态不可再跃迁，取消/成功/失败都受 WHERE 守卫
• 执行 Agent 逻辑的那几十秒不在任何事务里

这五条里，最后一条最容易被违反、代价也最大。它其实总结了整节的哲学：事务只用来保护「状态的原子跃迁」，不用来保护「业务逻辑的执行」。把这条守住，并发的正确性就只剩下「每条跃迁的 WHERE 写对」这一件事。

状态跃迁的原子性有了，但每次抢占都要扫一遍 runs 表找 queued 的行——数据一多，这个扫描就会变成瓶颈。下一节 5.3 查询优化：EXPLAIN 定位慢查询并加索引 用 EXPLAIN 把这类慢查询找出来并治掉。

阅读导航：上一节：5.1 Schema 设计：会话/消息/运行/记忆表 + 迁移 · 下一节：5.3 查询优化：EXPLAIN 定位慢查询并加索引 。