33 lines
1.3 KiB
Python
33 lines
1.3 KiB
Python
"""入隧/使用命令的有限整事务锁冲突重试，不重放结果未知的连接错误。"""
|
|
|
|
import asyncio
|
|
import secrets
|
|
from collections.abc import Awaitable, Callable
|
|
from functools import wraps
|
|
from typing import ParamSpec, TypeVar
|
|
|
|
from sqlalchemy.exc import OperationalError
|
|
|
|
P = ParamSpec('P')
|
|
R = TypeVar('R')
|
|
|
|
|
|
def retry_mysql_lock_conflicts(operation: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
|
|
"""仅用于异常出口已 rollback 的完整业务命令；所有重试沿用原payload/key。"""
|
|
max_attempts = 3
|
|
|
|
@wraps(operation)
|
|
async def wrapped(*args: P.args, **kwargs: P.kwargs) -> R:
|
|
for attempt in range(max_attempts):
|
|
try:
|
|
return await operation(*args, **kwargs)
|
|
except OperationalError as exc: # noqa: PERF203 -- 有限整事务重试必须逐次捕获锁冲突
|
|
errno = exc.orig.args[0] if getattr(exc.orig, 'args', ()) else None
|
|
# 3572=NOWAIT冲突，1213=死锁，1205=锁等待超时；不重试断连/未知提交结果。
|
|
if errno not in {3572, 1213, 1205} or attempt == max_attempts - 1:
|
|
raise
|
|
await asyncio.sleep(0.05 * (attempt + 1) + secrets.randbelow(20) / 1000)
|
|
raise AssertionError('锁冲突重试未返回结果')
|
|
|
|
return wrapped
|