Outbox 的原子意图与重复投递

实现与取舍

代码示例

事务回滚不会留下孤立任务或事件

SQLite 内存事务演示；不包含真实队列投递器、幂等消费或分布式压测。

import sqlite3

db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE runs(id TEXT PRIMARY KEY);
CREATE TABLE outbox(event_id TEXT PRIMARY KEY, run_id TEXT NOT NULL);
""")
def submit(run_id, fail=False):
    with db:
        db.execute("INSERT INTO runs VALUES (?)", (run_id,))
        if fail:
            raise RuntimeError("crash before outbox")
        db.execute("INSERT INTO outbox VALUES (?, ?)", (run_id + ":created", run_id))
try:
    submit("r1", fail=True)
except RuntimeError:
    pass
print("after rollback:", db.execute("SELECT count(*) FROM runs").fetchone()[0],
      db.execute("SELECT count(*) FROM outbox").fetchone()[0])
submit("r2")
print("after commit:", db.execute("SELECT count(*) FROM runs").fetchone()[0],
      db.execute("SELECT count(*) FROM outbox").fetchone()[0])
db.close()

预期输出

after rollback: 0 0
after commit: 1 1

工程推演

场景

面试假设：提交 Agent 任务返回成功，偶发任务一直停在 queued。

设计决策

任务和待投递事件同事务提交，后台重试并扫描滞留记录。

验证目标

已提交任务可恢复，重复投递不创建第二个运行。

适用边界

Outbox 不会自动解决外部工具副作用的 exactly-once。

易错点

• 把双写包在 try/catch 就算可靠
• 依赖队列自动去重解决全部问题
• 长期事务里等待模型

参考资料

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

• AWS：Transactional outbox pattern
• AWS：Retry with backoff pattern

检查自己理解到哪一步

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

基础达标

能指出先写库和先发消息各自窗口。

中高级信号

给出事务 Outbox、稳定 ID 和幂等消费。

资深信号

能设计宕机点验证、滞留告警与安全重投。