做 AI 生成服务的时候，我接入了几个下游系统：生成完成后通知对方的业务系统，对方再触发下一步。
上线第二周，对方的业务系统发来一条消息："你们有几次回调没到，我们那边有单子卡在中间态了。"
我第一反应是查我自己的日志——发出去了，HTTP 200，对方也返回 200。对账之后发现，他们那边有三条记录重复了，一条缺失。那天之后我重写了整个回调模块。
第一个坑：对方返回 5xx，我方不能就这么算了
最初我的实现很朴素：发出 HTTP 请求，收到 200 就标记"已送达"，其他情况就丢弃。
问题在于一次网络抖动就够了。对方的接口偶尔因为连接池满返回 503，重试一次就好的那种情况，我的实现直接把它当失败扔了。而且失败之后没有任何地方能看到。
| 改完之后的核心原则是："已送达"必须是对方明确确认，而不是我没收到错误。 具体做法是引入三态： | 状态 | 含义 | 后续动作 |
|---|---|---|---|
| delivered | 对方明确返回 2xx | 终态，不再管 | |
| retrying | 网络错误、超时、5xx、429 | 进重试队列 | |
| dead | 重试耗尽 | 进死信，人工介入 |
退避要带抖动。 固定间隔重试在对方故障恢复那一刻会惊群——几百个请求同时砸过去，把刚恢复的服务再打死。指数退避加随机抖动是必须的：
delay = min(base * (2 ** attempt), MAX_DELAY) * random.uniform(0.5, 1.5)4xx 和 5xx 要区别对待。 400、401、422 这类重试一万次也不会变对，只有 429 和 5xx 该重试。401 可能是密钥过期了，应该告警而不是重试。
必须有死信队列。 重试耗尽之后不能静默丢弃。我见过更糟的做法是重试次数无限，一个坏 payload 无限循环把队列塞满。
第二个坑：重复回调比丢回调更难处理
对方收到回调之后会至少重试一次——因为他们也不确定我这边的 200 是不是真返回了。所以你的回调端必须假设同一个事件会被收到多次。
我的处理分两层。第一层是去重，靠唯一键，事件 ID 由发送方生成：
# 发送方
event = {
"id": uuid4().hex, # 事件唯一 ID
"type": "generation.completed",
"created_at": int(time.time()),
"data": {...},
}接收方用这个 ID 做去重，靠数据库唯一约束，而不是先查再插：
CREATE TABLE processed_events (
event_id TEXT PRIMARY KEY,
event_type TEXT NOT NULL,
processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);def already_processed(event_id: str) -> bool:
db.execute(
"INSERT INTO processed_events (event_id, event_type) "
"VALUES (%s, %s) ON CONFLICT DO NOTHING",
(event_id, event_type),
)
return db.rowcount == 0 # 冲突 = 已经处理过先 SELECT 再 INSERT 会有竞态：两个回调同时到达，两个都查到"没处理过"，然后都执行了业务逻辑。ON CONFLICT DO NOTHING 让数据库来判断，天然原子。
第二层是幂等。去重表可能过期清理，幂等是最后一道防线。 两者解决的是不同问题：去重回答"这个事件我处理过吗"，幂等回答"这个业务操作我执行过吗"。对方可能对不同的事件 ID 发来同样的业务意图（重试换了 ID，或重放了整个流程），所以业务逻辑本身必须幂等。我给业务操作加了独立于事件 ID 的幂等键：
def on_generation_completed(event):
if already_processed(event["id"]):
return
# 业务操作用自己的幂等键，跟事件 ID 分开
key = f"delivery:{event['data']['order_id']}"
with db.transaction():
if redis.set(key, "1", nx=True, ex=86400):
actually_ship_the_order(event["data"])
# 已经发过货就直接跳过，接口照样返回 200
db.execute(
"INSERT INTO processed_events (event_id, event_type) VALUES (%s, %s) "
"ON CONFLICT DO NOTHING", (event["id"], event["type"]))注意顺序：先做业务，再记已处理。反过来会有一个问题——业务执行了但记录没写，下次重试又执行一遍。放在同一个事务里最稳。而且不管是否重复，都要返回 200，否则会触发对方的无意义重试。
第三个坑：签名验证挡不住重放
回调接口不验签名等于敞开。常见做法是对请求体做 HMAC：
import hmac, hashlib
def verify_signature(raw_body: bytes, header_sig: str, secret: str) -> bool:
expected = hmac.new(
secret.encode(), raw_body, hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected, header_sig)两个细节：
必须用原始字节验签。 先 JSON 解析再重新序列化，键顺序、空格、Unicode 转义都会变，签名对不上。要在框架解析 body 之前拿到 raw body。
用 compare_digest 而不是 ==。 普通字符串比较会在第一个不同的字节处返回，攻击者能靠测量响应时间逐字节猜出签名。
但光有签名还不够。攻击者可以截获一个合法的回调原样重放——签名对，事件却是三天前的。所以要加时间戳和重放窗口：
def verify(raw_body, header_sig, header_ts, secret, tolerance=300):
ts = int(header_ts)
if abs(time.time() - ts) > tolerance:
return False
# 时间戳要参与签名，否则可以随便改
expected = hmac.new(
secret.encode(),
f"{ts}.".encode() + raw_body,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected, header_sig)时间戳必须参与签名计算，否则攻击者改了时间戳就绕过了窗口检查。
窗口取 5 分钟是个权衡：太小会因为时钟漂移拒掉合法请求，太大会扩大重放窗口。前提是两边机器时间要同步。
顺序问题：回调不保证顺序
对方系统里 A 和 B 两个事件，A 先发 B 后发，但因为重试和网络乱序，B 先到了我的系统，B 触发的操作找不到 A 的中间状态，直接失败。
解决办法是给事件加个单调递增的序列号或版本号：
if event["seq"] <= last_seen_seq.get(event["type"], 0):
# 乱序或重复的旧事件，直接确认掉
return但这只解决"整体乱序"。真正的因果关系要靠事件带依赖信息，比如 depends_on: [event_a_id]，处理 B 之前检查 A 处理过没有。
一个容易被忽略的运维细节
回调发送要用独立的 HTTP 客户端配置：独立的连接池（否则业务流量会把回调挤掉）、较短的独立超时，以及不要用会自动重试的客户端库——很多库默认重试 3 次，我上面写的重试逻辑就变成了 3×3 次。
session = httpx.Client(
timeout=httpx.Timeout(10.0, connect=3.0),
limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
# 绝不在这里加 retries=，重试由队列逻辑统一控制
)还有：回调里不要同步等下游的下游，耗时的部分丢进队列。我见过一个回调接口同步调了三个外部系统，整体耗时 8 秒，对方的重试逻辑就疯狂触发。
我现在检查回调可靠性的方式
对账。 每天定时比对我方的"已发送"和对方的"已接收"，对不上的走人工。回调系统永远会有漏网的，对账是最后的兜底。
混沌演练。 把队列里的回调标记为失败，看重试是不是真在工作；重放一个已处理的事件，看幂等是不是真挡住了。
监控四个数： 投递成功率、重试次数分布、死信数量、处理 P99 延迟。死信数量大于 0 就该有人看一眼。