幂等键存储与重复请求重放

功能说明

IdempotencyStore 用于让可能重试的写操作具备幂等性。调用方为每次请求提供幂等键与载荷，存储先计算载荷的 SHA-256 指纹；若键存在且未过期，就比较指纹：一致则直接返回此前记录的响应，不一致则抛出 IdempotencyConflictError，防止同一个键被用于两笔不同业务。键不存在或已过期时执行 produce()，把响应与指纹、时间一起写入内存字典。TTL 通过注入的时钟惰性判断，读取时才删除过期条目。默认时钟是 time.monotonic，避免系统时间回拨影响过期计算。

关键代码

重放与冲突判断发生在 execute 的前半段：

fingerprint = payload_fingerprint(payload)
existing = self._get_live(key)
if existing is not None:
    if existing.payload_hash != fingerprint:
        raise IdempotencyConflictError(
            f"key {key!r} was already used with a different payload"
        )
    return existing.response
response = produce()

_get_live 在超出 ttl_s 时删除条目并返回 None，使过期后的同一键可以重新执行。空键在入口处即被拒绝，构造时还会校验 TTL 必须为正数。

使用注意

• 存储完全在单进程内存中，重启即丢失，多实例部署无法共享，生产环境应换成 Redis 等带原子操作的存储。
• 没有并发控制，两个线程同时用同一键调用可能都会执行 produce()，幂等性无法保证。
• 记录的是响应对象引用，若调用方修改返回值，后续重放会拿到被改动的数据。
• 指纹基于原始字符串，同一语义但字段顺序不同的 JSON 会被视为不同载荷。
• 未记录进行中的状态，produce() 抛错时不会留下条目，重试会再次执行。

验证方式

源码测试覆盖四种场景：首次执行后重放不再调用 produce、同键不同载荷抛冲突、推进假时钟超过 TTL 后重新执行、空键被拒绝。把实现保存为 example.py 并运行 python -m unittest 即可复现。手工验证可连续两次传入相同键与载荷，统计 produce 调用次数应为一。