自动化流程跑通用了两小时，稳定用了三周

周一早上打开日报，同一笔 398 元的订单在里面出现了三次，另外两笔该出现的完全没有。我翻日志，前一晚的流程确实「成功」了——退出码 0，最后一行写着 done。

这条流程不复杂：每晚拉支付后台、工单系统、CRM 三个来源的数据，按订单号对齐，写进一张汇总表，第二天八点半推一份摘要到群里。四个步骤，两百来行 Python。从有想法到跑通，我大概花了两小时；让它连续两周不出事，前前后后花了将近三周。踩的坑没有一个是业务逻辑本身的。

跑通那天，我以为这事就完了

第一版是个加长版脚本：

def main():
    orders = fetch_payments()
    tickets = fetch_tickets()
    customers = fetch_customers()
    rows = merge(orders, tickets, customers)
    write_table(rows)
    notify(rows)

cron 挂在 23:30，抛异常就退出，第二天再说。演示的时候它毫无问题，因为演示只跑一次，而且数据和网络都配合。真实环境里它依赖三个我一开始没意识到的前提：每次调用都会成功；每次成功的调用只发生一次；每次失败都可以整条重来。三条没有一条成立。

200 不等于成功，重试也不等于安全

支付后台在 token 过期时返回的是 HTTP 200，body 里是 {'code': 40101, 'msg': 'token expired'}，data 是空数组。我只判断了 resp.status_code == 200，于是拿到一个空列表继续往下跑，最后生成一份「当日无订单」的日报，我还给它点了赞。

后来所有外部调用都过一层封装：

def call(path, **params):
    resp = session.get(base + path, params=params, timeout=15)
    resp.raise_for_status()
    body = resp.json()
    code = body.get('code')
    if code != 0:
        raise UpstreamError(f'{path} code={code} msg={body.get("msg")}')
    return body['data']

关键不在于封装本身，而在于业务码的判断要收在入口，不能散落在每个调用点。这版是我事后补的，补之前已经有两处漏判，其中一处跑了三天才被发现。

重试是第二个坑。我给 call 加过「失败重试三次」的装饰器，第二天表里一堆重复行——写库那步失败过一次，重试时整条流程从第一步重新跑，导出又写了一遍。

装饰器没错，错在重试的粒度选在了整条流程上。现在的规矩是：超时、连接重置、429、5xx 在单次调用上退避重试，最多三次；其余 4xx 直接抛出去，参数错了重试一百次也还是错的。

而重复写入这件事，靠重试策略防不住，只能靠写入端兜底：

insert into daily_orders (order_id, amount, paid_at, synced_at)
values (%s, %s, %s, now())
on conflict (order_id) do update
  set amount = excluded.amount,
      paid_at = excluded.paid_at,
      synced_at = now();

业务主键必须是上游真实存在的东西，不能是「这一批的第 N 行」。我第一版用的自增 id，重复数据进来时毫无阻力，一路绿灯。

跨系统的流程没有事务，只能记进度

第三步写库成功、第四步发消息失败，数据库里已经是新数据，群里还是旧日报。跨三个外部系统不存在回滚这回事。我一度想用 try/except 把前三步包起来做补偿，写到一半放弃了——补偿逻辑比正向逻辑还长，每加一个来源就得重写一遍。

后来换成一个很土的办法：记进度。

create table flow_run (
  run_id      text not null,      -- 日期 + 流程名
  step        text not null,
  status      text not null,      -- running / done / failed
  payload     jsonb,
  updated_at  timestamptz default now(),
  primary key (run_id, step)
);

每一步开始前写 running，成功后改成 done，顺便把中间结果塞进 payload。重跑时从第一个非 done 的步骤开始。这样「整条重跑」就变成了「从断点续跑」，重复写入的问题少了一大半。

代价是步骤之间要能序列化。汇总那一段我直接存 jsonb，一天几千行，够用。如果中间数据涨到几十兆，这个方案就得换成落文件加路径引用，这一点我还没有实际验证。

还有一个更土但更有用的东西：单步可执行。

python -m flow run --step=export  --date=2026-03-01 --dry-run
python -m flow run --step=publish --date=2026-03-01

有了它，出问题时我不用等到第二天晚上才能验证，也不用在 notebook 里复制粘贴半条流程。--dry-run 只打印将要写入的行数和前五行样例，不落库。

这套东西加起来并不优雅。我一度想把它抽成一个「通用工作流引擎」，配置驱动、步骤可插拔。看了半天自己画的图，发现真实需求只有一条流、四个步骤，就停了。什么时候有第二条流再说。

告警要么太吵，要么太安静

第一版告警是「任何异常都发群」。上线第一周群里被刷了四十多条，第二周我把通知静音了。第三周流程静默挂了三天没人发现——因为它连异常都没抛，cron 那台机器磁盘满了，任务根本没启动。

这件事之后改了两点。

一是分级。真正需要当晚有人处理的，只有「会影响第二天八点半那份日报」的失败，其余先记流水，第二天早上汇总成一行。判断标准写成配置放在流程里，不靠当班的人临时判断。收敛窗口我定的是同类错误 30 分钟只发一次，代价是偶尔会漏掉第二次不同的失败现场，这个损失我认。

二是心跳。异常告警只能覆盖「跑失败了」，覆盖不了「根本没跑」。每天 09:30 查一次：

select count(*) from flow_run
where step = 'publish'
  and status = 'done'
  and updated_at > now() - interval '26 hours';

结果为 0 就告警。这条检查挂在另一台机器上，避免和主流程一起死掉。26 小时这个数留了两个小时的余量，再紧就会在周末误报。

日志也返工过一次。最早我把每次请求的完整 payload 打进日志，方便排查。某天发现日志里有用户的手机号和收货地址，而那份日志被同步到了共享盘。现在的规矩是：日志里只出现 id、状态码、行数、耗时，以及 payload 的 sha256 前八位。要复现问题就拿 id 去源头查，不把数据搬来搬去。

现在每起一条新流程，我先写三样东西：一张 flow_run 表、一个 --dry-run 开关、一条心跳查询。业务代码最后写。这个顺序听着别扭，但省下的是后面几周的排查时间，以及那种「明明跑成功了啊」的自我怀疑。

如果你手上也有一条正在跑的自动化流程，今晚可以做一件小事：去确认它上一次成功是什么时候，以及如果它今天没跑，你多久会知道。第二个问题答不上来，就先补那条心跳查询，别急着加功能。