一个幂等键只能在服务定义的作用域与保留窗口内标记一次逻辑行动。重复同一段字符串，不是放之四海皆准的“恰好执行一次”保证。服务需要把身份关联到正确的主体与操作，协调竞争中的请求，并保存足够权威的信息，以便稍后认出已经处理的行动。键只是入口，真实保证取决于这条处理链。
作用域是身份的一部分 如果全局只按客户端提交的键查找，就可能把不同租户的行动合并，甚至把别人的已存结果返回给当前调用方。查找或重放之前仍须完成认证与授权。应把可信作用域、稳定的操作标识、客户端行动键一起作为查找身份，并保留各组件边界，避免直接拼接字符串产生歧义。
究竟哪些维度属于作用域，要由 API 契约决定。EC2 明确区分地域级与可用区级幂等；按其文档规则，相同客户端令牌在不同地域或可用区可能允许执行不同的行动。因此，把重试切换到另一个地域，并不自动保留全局去重保证，不能只检查令牌文本是否一致。
请求含义也需要约束。同一个作用域内，原键配上不同参数时，必须按照已经声明的冲突规则处理。下面的实验只对一个正整数量生成比较表示，不比较任意 HTTP 正文，不执行认证，也不认为对正文取个哈希，就能自动推断业务语义是否相同。
让业务效果与结果记录一起提交 先查记录、再独立写入，会出现竞争：两个工作进程都没看见旧记录，于是都执行了业务。如果两份业务效果已经发生，之后只有一个结果记录插入成功，那么结果表上的唯一键本身并不能消除之前的重复效果。需要保护的是业务修改与记忆结果之间的完整边界。
两者都位于同一个事务数据库时，可以放在一个事务中提交。示例先执行 SQLite BEGIN IMMEDIATE，再查询记录，并在同一事务里写入业务行与结果记录。SQLite 同时只允许一个写事务，因此竞争连接会被串行处理；复合主键也明确表达了去重所依赖的身份组成。
事务保护止于这个数据库。外部支付、邮件发送或消息投递，不会因为本地事务失败就自动撤销；这些效果需要自己的端到端投递与去重契约。把整个流程命名为“恰好一次”，并不能凭空补出跨越多个系统的原子性，评审时必须沿真实副作用检查边界。
保留期也是正确性边界 如果旧请求仍可能回来，而记住它的键已经过期，那次请求就可能重新执行。Stripe 文档明确说明，记录清理后的同键请求会成为新请求；其至少二十四小时的保留说明属于该提供方，并不是所有服务通用的计时器，更不能解释成永久去重。
恢复窗口应与保留期一致，还要区分临时去重记录和长期业务唯一性规则。删除结果记录，不会撤销原来的订单。如果业务禁止这笔订单在任何时候被重新创建，就需要一个比短期重试缓存更持久的权威约束，不能把希望都放在键还没过期这件事上。
恢复较旧的记录数据库、清空状态或切换到独立存储，也可能忘记已经完成的行动。恢复方案应保留业务数据与操作记录之间的关系，不能只证明各自的文件还能读取，却没有验证它们是否对应同一批已提交事实。
坏掉的样子 提交成功后丢失响应，会让客户端不知道结果，但不会自动删除已存记录。提交前事务失败，则应同时不留下业务效果与成功记录。这是两种不同情况，需要不同证据，不能把“客户端看到了错误”直接等同于“服务器没有执行”。
数据库竞争、存储不可用或未知结果，也不能被当成“查不到旧记录”。看到操作仍在执行，不应自动放行第二份业务效果。需要明确重复请求是等待、得到执行中响应，还是走另一条已有文档的恢复路径；成功结果重放只是可能状态之一，不覆盖所有中间阶段。
快速验证 保存为 scope_check.py，用 Python 3.10 以上版本执行 python3 scope_check.py。实验已在 Python 3.14.4、SQLite 3.53.0 上运行。标准库代码在私有临时目录创建数据库，结束后清理；每次尝试都打开自己的连接。租户与操作参数代表已经完成身份验证和路由解析的测试输入，不是客户端请求头的替代品。
from concurrent.futures import ThreadPoolExecutor from contextlib import closing import json from pathlib import Path import sqlite3 from tempfile import TemporaryDirectory from threading import Barrier with TemporaryDirectory(prefix="idempotency-scope-") as directory: database = Path(directory) / "orders.sqlite" with closing(sqlite3.connect(database)) as db: db.executescript(""" CREATE TABLE effects ( id INTEGER PRIMARY KEY, tenant TEXT NOT NULL, operation TEXT NOT NULL, quantity INTEGER NOT NULL CHECK (quantity > 0) ); CREATE TABLE receipts ( tenant TEXT NOT NULL, operation TEXT NOT NULL, request_key TEXT NOT NULL, fingerprint TEXT NOT NULL, result TEXT NOT NULL, expires_at INTEGER NOT NULL, PRIMARY KEY (tenant, operation, request_key) ); """) def execute(tenant, operation, key, quantity, now, fail_before_receipt=False): # Tenant/operation are already authenticated/routed fixture identifiers. assert type(quantity) is int and quantity > 0 fingerprint = json.dumps({"quantity": quantity}, separators=(",", ":")) scope = (tenant, operation, key) with closing(sqlite3.connect(database, timeout=5)) as db: with db: db.execute("BEGIN IMMEDIATE") previous = db.execute( "SELECT fingerprint, result, expires_at FROM receipts " "WHERE tenant=? AND operation=? AND request_key=?", scope, ).fetchone() if previous is not None and previous[2] > now: if previous[0] != fingerprint: return "conflict", None return "replay", json.loads(previous[1]) if previous is not None: db.execute("DELETE FROM receipts WHERE tenant=? AND operation=? AND request_key=?", scope) cursor = db.execute( "INSERT INTO effects (tenant, operation, quantity) VALUES (?, ?, ?)", (tenant, operation, quantity), ) result = {"effect_id": cursor.lastrowid, "quantity": quantity} if fail_before_receipt: raise RuntimeError("injected failure after effect insertion, before receipt insertion") db.execute("INSERT INTO receipts VALUES (?, ?, ?, ?, ?, ?)", (*scope, fingerprint, json.dumps(result), now + 60)) return "created", result def counts(): with closing(sqlite3.connect(database)) as db: return db.execute("SELECT (SELECT count(*) FROM effects), (SELECT count(*) FROM receipts)").fetchone() gate = Barrier(8) def attempt(_): gate.wait(timeout=5) return execute("north", "web-order", "intent-a", 2, 0) with ThreadPoolExecutor(max_workers=8) as pool: outcomes = list(pool.map(attempt, range(8))) assert [kind for kind, _ in outcomes].count("created") == 1 assert [kind for kind, _ in outcomes].count("replay") == 7 original = outcomes[0][1] assert all(result == original for _, result in outcomes) and counts() == (1, 1) print("eight independent connections produce one effect, one receipt, and seven replays") assert execute("north", "web-order", "intent-a", 3, 1) == ("conflict", None) assert counts() == (1, 1) assert execute("south", "web-order", "intent-a", 2, 1)[0] == "created" assert execute("north", "batch-order", "intent-a", 2, 1)[0] == "created" assert execute("north", "web-order", "intent-b", 2, 1)[0] == "created" assert counts() == (4, 4) print("changed parameters conflict; a different tenant, operation, or intent key has separate scope") try: execute("north", "web-order", "rolled-back", 1, 2, fail_before_receipt=True) except RuntimeError as error: assert str(error).startswith("injected failure") else: raise AssertionError("expected the transaction to fail") assert counts() == (4, 4) assert execute("north", "web-order", "rolled-back", 1, 2)[0] == "created" assert counts() == (5, 5) print("a raised exception rolls back the effect and receipt together; retry then creates one effect") _, committed = execute("north", "web-order", "committed-no-reply", 1, 3) # Discard the reply, then query through a fresh connection; no network loss is simulated here. assert execute("north", "web-order", "committed-no-reply", 1, 3) == ("replay", committed) assert counts() == (6, 6) print("a fresh connection replays a committed result without adding another effect") assert execute("north", "web-order", "intent-a", 2, 59) == ("replay", original) kind, after_expiry = execute("north", "web-order", "intent-a", 2, 60) assert kind == "created" and after_expiry["effect_id"] != original["effect_id"] assert counts() == (7, 6) print("at the explicit 60-second retention boundary, the same key creates a new effect") print("SQLite", sqlite3.sqlite_version, "verified counts: 7 effects, 6 retained receipts") 八条竞争连接最终产生一条业务记录、一条结果记录和七次重放，所有返回结果一致。改变数量会冲突；改变租户、操作或行动键，则按已经声明的作用域创建独立工作，使计数达到四条业务记录与四条结果记录。
...
