队列的投递承诺只描述工作流程中的一段边界。消息到达消费者，与业务副作用已经提交，是不同事件。至少一次允许重复投递，至多一次则允许在不确定时丢失而不继续重试。仅凭这两个标签，不能保证订单、通知或数据库更新在所有故障下都恰好完成一次。可靠性需要沿真正的数据和确认路径逐步检查。
找到发布、确认与提交的位置 首先明确生产方何时知道 broker 已经接受消息、保留多久，以及持久化政策覆盖哪些故障。消息从未成功发布、超过保留期或被明确删除，都不会因为消费者自称至少一次就自动恢复。恢复前提必须与投递标签一起记录，否则“永不丢失”会遮住多个根本没有交给队列负责的环节。
消费方若先确认，再提交业务结果，就会形成丢失窗口：broker 停止重试时，应用可能还没完成工作。先提交又留下另一种窗口：结果已成功，但确认丢失，工作于是重新出现。这个重复仍然符合投递契约，并不意味着最初操作失败。排查时要分别检查业务结果与确认结果，不能用其中一个推断另一个。
可见性超时与确认期限控制的是何时允许下一次尝试，不证明第一个工作进程已经停止。慢处理者可以与重投后的处理者同时运行，因此延长期限与防止重复副作用解决的是不同问题。应同时记录投递尝试和已经提交的业务结果，而不是只统计消费者调用次数，便把它当作成功执行次数。
在真正的事务边界内消除重复效果 若副作用都位于同一个事务数据库，可以把唯一操作身份与业务结果放进同一事务。唯一性检查必须能够裁决并发尝试，单独读取再无保护地写入并不够。还需要保存足够的载荷信息，识别“同一身份却携带不同指令”的情况，不能看到编号重复就无条件接受。这既是去重问题，也是操作身份的契约问题。
数据库事务不会自动包住外部邮件、支付接口或另一套独立存储。这些目标需要自己的幂等或协调契约。本地标记先写、远端调用后做，可能丢失效果；标记后写，则可能留下重复窗口。因此，本地去重表不能被说成端到端恰好一次保证。只有明确哪些动作一起提交、哪些动作仍独立失败，设计才有可检查的边界。
有些服务确实提供带明确范围的恰好一次投递能力，例如 Google Pub/Sub 说明了地域、订阅类型等限制，也区分投递重复与发布侧重复。应阅读具体契约，不应宣称这个能力不可能存在，也不能把它推广成所有下游副作用都已解决。多个已发布消息代表同一业务意图时，稳定业务身份仍然有独立价值。
坏掉的样子 持续失败的消息可能阻塞有序进度，过短期限则可能让本来正常的任务出现重叠尝试。重投计数能提供调查线索，却不能唯一判断是毒数据、慢依赖、确认丢失还是工作进程停滞。应关联消息身份、载荷、异常、处理时长和确认记录，再决定怎样修改重试政策。只根据“一条还是多条计数升高”断言根因，很容易修错位置。
需要限制重试并说明隔离失败如何审阅或重新执行。进入死信队列只是移动未完成工作，并没有完成它。去重记录的保存期限必须覆盖计划容忍的重放范围，删除身份记录之后，旧请求可能再次生效；永远保存又有存储和隐私代价。这里没有适用于所有任务的统一选择，应把保留与重放规则一起设计。
历史精确 SQLite 落点证据涵盖回滚、重开和去重。总计是虚构夹具值；broker 行为与外部效果仍是独立边界。 快速验证 保存为 delivery_guarantee_probe.py，用启用断言的 Python 3 运行，不能使用 -O 或 PYTHONOPTIMIZE，因为断言中含有必需调用。保留的精确代码块记录报告 Python 3.12.14 与 SQLite 3.53.1 上通过；本轮没有重跑。临时 SQLite 数据库同时保存身份与无害数字总计，检查的是事务落点，不会扣款或发送外部消息，创建及清理的只是本程序自己的临时文件。
from contextlib import closing from pathlib import Path from tempfile import TemporaryDirectory import sqlite3 def apply(db, message_id, amount, interrupt=False): db.execute('BEGIN IMMEDIATE') try: prior = db.execute('SELECT amount FROM inbox WHERE id=?', (message_id,)).fetchone() if prior is not None: if prior[0] != amount: raise ValueError('same identity, different payload') db.commit() return False db.execute('INSERT INTO inbox VALUES (?, ?)', (message_id, amount)) db.execute('UPDATE totals SET amount=amount+? WHERE id=1', (amount,)) if interrupt: raise RuntimeError('controlled failure before commit') db.commit() return True except BaseException: db.rollback() raise with TemporaryDirectory(prefix='queue-sink-') as directory: path = Path(directory) / 'sink.db' with closing(sqlite3.connect(path, isolation_level=None)) as db: db.execute('CREATE TABLE inbox(id TEXT PRIMARY KEY, amount INTEGER NOT NULL)') db.execute('CREATE TABLE totals(id INTEGER PRIMARY KEY, amount INTEGER NOT NULL)') db.execute('INSERT INTO totals VALUES (1, 0)') try: apply(db, 'work-1', 100, interrupt=True) except RuntimeError: pass else: raise AssertionError('controlled failure did not occur') assert db.execute('SELECT count(*) FROM inbox').fetchone() == (0,) assert db.execute('SELECT amount FROM totals').fetchone() == (0,) assert apply(db, 'work-1', 100) assert apply(db, 'work-2', 50) # Close and reopen the real database rather than retaining an in-memory set. with closing(sqlite3.connect(path, isolation_level=None)) as db: assert not apply(db, 'work-1', 100) try: apply(db, 'work-1', 999) except ValueError: pass else: raise AssertionError('conflicting payload reused the same identity') assert db.execute('SELECT count(*) FROM inbox').fetchone() == (2,) assert db.execute('SELECT amount FROM totals').fetchone() == (150,) assert not db.in_transaction print('rollback kept marker and effect together; reopened retry changed nothing; total=150') 两项写入之后主动抛出的异常，会让身份标记与总计一起回滚。随后重试，提交两项操作；关闭并重新打开数据库后，重复第一项身份不会改变结果，而用不同数额复用相同身份会报错。最终数据库有两个身份，总计为一百五十，并且连接不在事务中。这个检查比只把编号放进内存集合更能说明持久化边界。
...
