工具超时后重试，如何防止重复发信、重复下单或重复发布？

实现与取舍

代码示例

SQLite 事务内的业务操作去重

Python 3 标准库演示，使用内存 SQLite；真实恢复需要文件或服务端数据库，外部动作需要远端幂等能力。

import hashlib
import json
import sqlite3

db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE publications (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    tenant TEXT NOT NULL,
    title TEXT NOT NULL
);
CREATE TABLE operations (
    tenant TEXT NOT NULL,
    op_key TEXT NOT NULL,
    payload_hash TEXT NOT NULL,
    result_id INTEGER NOT NULL,
    PRIMARY KEY (tenant, op_key)
);
""")

def publish(tenant, op_key, title):
    payload = json.dumps({"title": title}, sort_keys=True).encode()
    digest = hashlib.sha256(payload).hexdigest()
    db.execute("BEGIN IMMEDIATE")
    try:
        old = db.execute(
            "SELECT payload_hash, result_id FROM operations WHERE tenant=? AND op_key=?",
            (tenant, op_key),
        ).fetchone()
        if old:
            if old[0] != digest:
                raise ValueError("IDEMPOTENCY_CONFLICT")
            db.commit()
            return old[1]
        cur = db.execute(
            "INSERT INTO publications(tenant,title) VALUES (?,?)", (tenant, title)
        )
        result_id = cur.lastrowid
        db.execute("INSERT INTO operations VALUES (?,?,?,?)",
                   (tenant, op_key, digest, result_id))
        db.commit()
        return result_id
    except Exception:
        db.rollback()
        raise

print(publish("t1", "task-7:publish:v3", "Agent recovery"))
print(publish("t1", "task-7:publish:v3", "Agent recovery"))
print("rows", db.execute("SELECT COUNT(*) FROM publications").fetchone()[0])
try:
    publish("t1", "task-7:publish:v3", "Changed title")
except ValueError as exc:
    print(str(exc))
db.close()

预期输出

1
1
rows 1
IDEMPOTENCY_CONFLICT

工程推演

场景

假设工程场景：文章发布 API 超时，Agent 恢复后重新请求发布。

设计决策

操作键绑定任务与草稿版本；同键重试返回原发布结果；参数变化要求新审批与新操作。

验证目标

演示结果：两次相同调用返回同一 ID，数据库只有一条发布记录，不同参数被拒绝。

适用边界

代码只证明单数据库事务内去重；不证明外部网络动作 exactly-once，也未实现分布式投递。

易错点

• 把本地 timeout 当作远端未执行
• 先查后写但没有事务和唯一约束
• 认为保存检查点即可保证外部副作用只发生一次

参考资料

依据公开技术资料设计；参考资料支持技术机制，场景与评分标准为本站设计，不代表某公司面试原题。 新增问答与迁移案例用于原理讲解，来源核查与案例运行验证分别记录。

• LangGraph：Interrupts 与副作用重放
• Python：sqlite3 事务与参数绑定
• Stripe：幂等请求的参数比较与保留期限
• AWS：通过幂等 API 安全重试

检查自己理解到哪一步

读完后可以对照这些标准解释原理、边界和取舍。掌握程度由你自评；需要进一步验证时，再完成下方小任务。

基础达标

知道超时可能发生在外部提交之后。

中高级信号

使用稳定业务幂等键、意图账本和回执核对。

资深信号

明确不支持幂等时的未知终态、人工处理与故障注入。