21 · 事件流与 Outbox：消除不可靠双写
目标：用事务内 Outbox 和消费端去重处理提交后崩溃与消息重投。
两次成功之间存在一条裂缝
报名写入数据库后直接发消息：数据库成功、消息失败时，日报永远漏一条。先发消息再写数据库：数据库失败时，日报却多一条。这不是顺序选择题，而是两个系统没有共同原子边界。
Outbox 把报名和待发送事件写进同一数据库事务，再由转发器异步发送。它保证已提交报名有对应的待发记录，不保证只发送一次。接收端成功而确认丢失时，转发器还会重发，因此接收端要把去重标记和业务效果放在同一事务。
传统任务队列倾向于逐条确认并分发工作；日志型代理保留消息和消费位置，适合重放与多个独立下游。同一消费者组分摊处理，不同组各自消费全部事件。分区内顺序不代表跨分区全局顺序，保留期也意味着落后太久可能无法直接追赶。
让接收成功，但发送端不知道
保存 event_delivery.py。本例使用两个独立 SQLite 文件模拟源与接收端，日志转发由函数代替网络；临时目录在结束时清理。先预测两次交付后的计数。
import sqlite3
from pathlib import Path
from tempfile import TemporaryDirectory
with TemporaryDirectory() as work:
source = sqlite3.connect(Path(work) / "source.db", isolation_level=None)
sink_path = Path(work) / "sink.db"
sink = sqlite3.connect(sink_path, isolation_level=None)
source.executescript("""
CREATE TABLE registration(id TEXT PRIMARY KEY, user_id TEXT);
CREATE TABLE outbox(id TEXT PRIMARY KEY, user_id TEXT, sent INTEGER DEFAULT 0);
BEGIN IMMEDIATE;
INSERT INTO registration VALUES ('r1', 'u1');
INSERT INTO outbox(id, user_id) VALUES ('r1', 'u1');
COMMIT;
""")
sink.executescript("""
CREATE TABLE seen(id TEXT PRIMARY KEY, user_id TEXT);
CREATE TABLE totals(value INTEGER);
INSERT INTO totals VALUES (0);
""")
def apply(event_id, user_id):
sink.execute("BEGIN IMMEDIATE")
try:
prior = sink.execute("SELECT user_id FROM seen WHERE id = ?", (event_id,)).fetchone()
if prior:
if prior[0] != user_id:
raise ValueError("event identity conflict")
else:
sink.execute("INSERT INTO seen VALUES (?, ?)", (event_id, user_id))
sink.execute("UPDATE totals SET value = value + 1")
sink.execute("COMMIT")
except Exception:
sink.execute("ROLLBACK")
raise
def relay(lose_ack=False):
for event_id, user_id in source.execute("SELECT id, user_id FROM outbox WHERE sent = 0 ORDER BY id").fetchall():
apply(event_id, user_id)
if lose_ack:
return
source.execute("UPDATE outbox SET sent = 1 WHERE id = ?", (event_id,))
relay(lose_ack=True)
assert source.execute("SELECT sent FROM outbox").fetchone()[0] == 0
sink.close()
sink = sqlite3.connect(sink_path, isolation_level=None)
relay()
total = sink.execute("SELECT value FROM totals").fetchone()[0]
pending = source.execute("SELECT COUNT(*) FROM outbox WHERE sent = 0").fetchone()[0]
assert total == 1 and pending == 0
print("count_after_redelivery:", total)
print("pending:", pending)
source.close()
sink.close()
uv run --python 3.12 --no-project event_delivery.py
预期输出：
count_after_redelivery: 1
pending: 0
第一次交付提交了接收端效果，但保留源端未发送状态。接收端关闭并重新连接后，去重表还存在，所以重发没有再加一。这是特定数据库效果的“处理一次”，不是消息只传一次，更不自动覆盖发邮件等外部效果。
引导练习：① 连续调用三次 relay(lose_ack=True)；② 再正常 relay()；③ 总数仍为 1。独立练习：添加 r2/u2 的报名与 Outbox，制造同样确认丢失，最终检查计数为 2、去重行数为 2、待发为 0。
参考修改
在源数据库初始化事务中增加 INSERT INTO registration VALUES ('r2', 'u2'); 与 INSERT INTO outbox(id, user_id) VALUES ('r2', 'u2');。将末尾总数断言改为 2，增加 assert sink.execute('SELECT COUNT(*) FROM seen').fetchone()[0] == 2。即使首次转发在 r1 后返回，后续正常转发也会把 r1 去重并处理 r2。
CDC 从数据库已提交变化中提取变更，可用于转发 Outbox，也可直接传播行变化；事件溯源则把业务事件本身作为事实源，两者不等价。CDC 初始化要把一致快照和日志位置衔接，重建还要确定日志是否完整。积压时看最老未处理事件年龄和保留期，不能只看消息条数。
若重投翻倍，检查去重与加一是否同事务；若永久漏消息，检查是否先标记 sent 再处理。多个下游需要各自的交付位置，示例一个 sent 标记只适用于这一个接收目标。将此边界与第 14 课事务合并写入 design.md。