Python后端中间件专题07：用户点两次提交怎么只建一张单——幂等键与请求指纹

Python后端中间件专题07：用户点两次提交怎么只建一张单——幂等键与请求指纹

客户端发出 POST 后没收到响应，无法知道是“服务端没执行”还是“已提交，回包丢了”。重试是正常行为，服务端必须让同一业务意图收敛到同一 Ticket ID。但“收到重复 key 就返回老结果”还不够：我们必须证明重试的是同一意图。

不要先问“这个 key 见过没有”，先让两个请求故意撞在同一 key 上。你会为请求生成稳定 SHA-256 指纹，区分同键同体 replay 和同键异体 conflict，并观察唯一约束如何让两个都没看见 receipt 的事务只留下一个 ticket 和一个 Outbox。

这需要第 02 篇的 201/409 契约、第 06 篇的准入拒绝，以及 SQLAlchemy 事务、savepoint、IntegrityError。本地文件 SQLite 测的是冲突恢复控制流；它不替代 PostgreSQL 隔离和锁可见性的证据。

上一课练习答案

先排除一个危险的误会：被拒绝不等于已提交。

答案 EX-06-01 [code]

defdecide(redis_ok:bool,endpoint:str)->dict[str,object]:ifendpointnotin{"create","detail"}:raiseValueError("unknown endpoint")ifredis_ok:return{"allowed":True,"status":200,"fallback":None}ifendpoint=="create":# 保护型写入无法证明预算时 fail-closed。return{"allowed":False,"status":503,"fallback":None}# 普通详情有 DB 连接预算托底时可有界 fail-open。return{"allowed":True,"status":200,"fallback":"db"}cases=[(True,"create",True,200,None),(True,"detail",True,200,None),(False,"create",False,503,None),(False,"detail",True,200,"db"),]forredis_ok,endpoint,allowed,status,fallbackincases:result=decide(redis_ok,endpoint)assert(result["allowed"],result["status"],result["fallback"])==(allowed,status,fallback)print("4 outage-policy cases passed")

运行命令：python outage_policy.py

预期输出：

4 outage-policy cases passed

生产中是返回 503 还是 429 要取决于你是否能区分“依赖不可用”与“已确认额度用尽”。TicketFlow 使用 429 表示已知预算耗尽，Redis 故障导致无法裁决则应暴露为依赖降级，不混淆两者。

答案 EX-06-02 [prose]

限流拒绝表示本次创建没有进入业务事务，因此没有 ticket ID 可以重放。幂等重放则表示数据库中已有(tenant_id, key, fingerprint) → ticket_id的持久收据，本次请求只是重取结果。把 429/503 伪装为幂等成功会生成一个不存在的业务事实；相反，对已提交请求再返限流失败又会迫使客户端无限重试。唯一可靠的分界是数据库收据。

先让两个事务撞到同一张 receipt 上

检查点用两个线程和 barrier 让它们都在首次 receipt SELECT 处观察到空，稳定重现 check-then-insert 竞态。最终断言两个返回 ID 相同，Outbox 只有一行。

运行命令：python -m pytest tests/unit/test_ticket_repository.py::test_concurrent_same_key_retries_converge_to_one_ticket -q

本地 Python 3.11 实际输出：

. [100%] 1 passed

此证据在文件 SQLite 上验证 repository 冲突恢复逻辑。可信远端 selector 会先用 PostgreSQL 表锁作为受控 barrier，证实 api-a/api-b 的两个事务同时在idempotency_receipts临界区等待，再释放锁让同租户、同 key、同 body 竞争；两个 201 必须返回同一 ID，并直接查询 PostgreSQL 的 ticket/receipt/created Outbox 各一。同键异体必须 409且计数不变，另一租户可独立使用同 key。未执行前仍为 PENDING。

从碰撞结果拆开“同 key”的两种含义

TicketFlow 将actor_id、title、body和priority用稳定 key 顺序、无多余空格的 JSON 序列化，然后计算 SHA-256。指纹不存明文请求不代表它是密码哈希：这里的目标是稳定比较意图，不是对低熵 body 保密。日志不应输出原始 body，数据库中存指纹与工单外键即可。

唯一性范围是(tenant_id, idempotency_key)。两个租户使用同一客户端 key 不应相互影响；同一租户内的 key 则必须唯一。这个约束要由 PostgreSQL 执行，不能用 RedisSET NX代替，因为 Redis 数据可过期、丢失或在 failover 窗口中出现不同观察。

结果为什么必须交给数据库，而不是先查再猜

两个事务可能同时查不到 receipt，都试图插入。不能靠“先 SELECT 了”保证安全，真正的仲裁点是唯一约束。repository 用 savepoint 包住 ticket、Outbox 和 receipt 的插入：输家收到IntegrityError后只回滚子事务，再查已提交的赢家 receipt。指纹相同就返原 Ticket，不同就抛冲突。

代码中工单、Outbox 事件和幂等收据在同一session.begin()中。这保证不会出现“有工单但没有可交付事件”或“有 receipt 但 ticket 未提交”的半状态。更新工单时 version 递增，并在同事务写ticket.updatedOutbox，为后面的缓存/搜索收敛提供可持久事件。

观察到两个线程仍收敛为同一 ID 后，反过来检查实现没有偷换前提：只比较 key 而不比较指纹会把另一业务意图误重放；用str(dict)会造成虚假冲突；先 SELECT 为空也不是安全插入；捕获唯一约束后回滚整个外层事务会破坏赢家 receipt 的回读；把 ticket、receipt、Outbox 分成三次事务又会制造半状态。它们共同说明，真正的仲裁是带规范指纹的数据库唯一约束和同事务写入，而不是临时 Redis 状态。

本课练习

现在给三次碰撞写出判词。

练习 EX-07-01 [code]

为下列三个请求写一个纯函数判定器和断言：租户 A/key-1/body-X 已有 receipt；再来 A/key-1/body-X 应是replay；A/key-1/body-Y 应是conflict；B/key-1/body-X 应是create。请使用规范 JSON 的 SHA-256 指纹，不要直接比较 dict 的字符串显示。

练习 EX-07-02 [prose]

说明为什么幂等 receipt 不能只存 Redis，以及在已有数据库唯一约束后，再在 Redis 上为同 key 加短租约最多只能优化什么、不能保证什么。

第 08 篇会回答 EX-07-01/02，并把 Redis 短租约放回它该在的位置：它只能减少竞争，不能替代 receipt 唯一约束。届时会用两个 token 重演旧持有者误删新租约的风险。

完整核心模块：事务型工单 repository

"""Transactional SQLAlchemy repository for tickets and their publishable changes."""from__future__importannotationsimporthashlibimportjsonfromdatetimeimportdatetime,timezonefromtypingimportCallable,Optionalfromuuidimportuuid4fromsqlalchemyimportselectfromsqlalchemy.excimportIntegrityErrorfromsqlalchemy.ormimportSessionfromticketflow.domain.ticketsimportIdempotencyConflict,Ticketfromticketflow.storage.modelsimport(IdempotencyReceiptModel,OutboxEventModel,TicketModel,)SessionFactory=Callable[[],Session]classTicketRepository:"""Keeps a ticket mutation and its eventual-delivery record in one database commit."""def__init__(self,sessions:SessionFactory,)->None:self._sessions=sessionsdefcreate_ticket(self,tenant_id:str,actor_id:str,title:str,body:str,priority:str,idempotency_key:str,)->Ticket:payload_hash=self._create_payload_hash(actor_id,title,body,priority)withself._sessions()assession:withsession.begin():receipt=session.scalar(select(IdempotencyReceiptModel).where(IdempotencyReceiptModel.tenant_id==tenant_id,IdempotencyReceiptModel.idempotency_key==idempotency_key,))ifreceiptisnotNone:ifreceipt.payload_hash!=payload_hash:raiseIdempotencyConflict("idempotency key was already used with a different payload")returnself._to_ticket(receipt.ticket)try:# The savepoint lets a concurrent receipt winner be replayed without# aborting the outer transaction (valid for PostgreSQL and SQLite).withsession.begin_nested():now=datetime.now(timezone.utc)ticket=TicketModel(id=str(uuid4()),tenant_id=tenant_id,actor_id=actor_id,title=title,body=body,priority=priority,status="open",version=1,created_at=now,updated_at=now,)session.add(ticket)event_id=str(uuid4())session.add(OutboxEventModel(id=event_id,aggregate_id=ticket.id,tenant_id=tenant_id,event_type="ticket.created",payload_json=self._event_payload(ticket,event_id,now,"ticket.created"),occurred_at=now,))session.add(IdempotencyReceiptModel(tenant_id=tenant_id,idempotency_key=idempotency_key,payload_hash=payload_hash,ticket_id=ticket.id,created_at=now,))session.flush()returnself._to_ticket(ticket)exceptIntegrityError:# The unique receipt is the cross-process arbitration point. Once the# losing savepoint rolls back, query the committed winner and replay it.receipt=session.scalar(select(IdempotencyReceiptModel).where(IdempotencyReceiptModel.tenant_id==tenant_id,IdempotencyReceiptModel.idempotency_key==idempotency_key,))ifreceiptisNone:raiseifreceipt.payload_hash!=payload_hash:raiseIdempotencyConflict("idempotency key was already used with a different payload")returnself._to_ticket(receipt.ticket)defget_ticket(self,tenant_id:str,ticket_id:str)->Optional[Ticket]:withself._sessions()assession:ticket=session.scalar(select(TicketModel).where(TicketModel.tenant_id==tenant_id,TicketModel.id==ticket_id))returnself._to_ticket(ticket)ifticketisnotNoneelseNonedefupdate_ticket(self,tenant_id:str,ticket_id:str,title:Optional[str]=None,body:Optional[str]=None,priority:Optional[str]=None,)->Optional[Ticket]:withself._sessions()assession:withsession.begin():ticket=session.scalar(select(TicketModel).where(TicketModel.tenant_id==tenant_id,TicketModel.id==ticket_id))ifticketisNone:returnNoneiftitleisnotNone:ticket.title=titleifbodyisnotNone:ticket.body=bodyifpriorityisnotNone:ticket.priority=priority ticket.version+=1ticket.updated_at=datetime.now(timezone.utc)event_id=str(uuid4())session.add(OutboxEventModel(id=event_id,aggregate_id=ticket.id,tenant_id=tenant_id,event_type="ticket.updated",payload_json=self._event_payload(ticket,event_id,ticket.updated_at,"ticket.updated"),occurred_at=ticket.updated_at,))returnself._to_ticket(ticket)@staticmethoddef_create_payload_hash(actor_id:str,title:str,body:str,priority:str)->str:canonical=json.dumps({"actor_id":actor_id,"body":body,"priority":priority,"title":title},separators=(",",":"),sort_keys=True,)returnhashlib.sha256(canonical.encode("utf-8")).hexdigest()@staticmethoddef_event_payload(ticket:TicketModel,event_id:str,occurred_at:datetime,event_type:str,)->str:returnjson.dumps({"event_id":event_id,"ticket_id":ticket.id,"tenant_id":ticket.tenant_id,"version":ticket.version,"trace_context":{},"occurred_at":occurred_at.isoformat(),"payload":{"event_type":event_type,"priority":ticket.priority,"status":ticket.status,"title":ticket.title,},},separators=(",",":"),sort_keys=True,)@staticmethoddef_to_ticket(ticket:TicketModel)->Ticket:returnTicket(id=ticket.id,tenant_id=ticket.tenant_id,actor_id=ticket.actor_id,title=ticket.title,body=ticket.body,priority=ticket.priority,status=ticket.status,version=ticket.version,created_at=ticket.created_at,updated_at=ticket.updated_at,)

id,
title=ticket.title,
body=ticket.body,
priority=ticket.priority,
status=ticket.status,
version=ticket.version,
created_at=ticket.created_at,
updated_at=ticket.updated_at,
)