54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
"""实验用本地事务处理器
|
|
|
|
随子服务器进程（runserver）启动即注册，供主服务器 Saga 编排回调。
|
|
msync_experiment 管理命令与 HTTP 服务共用同一进程内处理器。
|
|
"""
|
|
|
|
import uuid
|
|
|
|
from gvsdsdk.msync.local_tx import local_tx
|
|
|
|
# demo_flaky 重试实验状态（进程内计数器，可通过 local/reset 端点清零）
|
|
_flaky_state = {'attempts': 0}
|
|
|
|
|
|
def _ActionA(payload):
|
|
return {'demo_a_uuid': str(uuid.uuid4()), 'received': payload}
|
|
|
|
|
|
def _ActionB(payload):
|
|
# 返回收到的完整 payload，供主服务器验证步骤间数据流
|
|
return {'received_payload': payload}
|
|
|
|
|
|
def _CompensateA(payload, action_result):
|
|
return {'compensated': True, 'action_result': action_result}
|
|
|
|
|
|
def _ActionFail(payload):
|
|
raise RuntimeError('确定性失败（实验）')
|
|
|
|
|
|
def _ActionFlaky(payload):
|
|
_flaky_state['attempts'] += 1
|
|
if _flaky_state['attempts'] == 1:
|
|
raise RuntimeError('首次失败（实验：验证自动重试）')
|
|
return {'retried': True, 'attempts': _flaky_state['attempts']}
|
|
|
|
|
|
def RegisterExperimentHandlers():
|
|
"""注册全部实验处理器（幂等：已存在则跳过）"""
|
|
for name, action, compensate in (
|
|
('demo_a', _ActionA, _CompensateA),
|
|
('demo_b', _ActionB, None),
|
|
('demo_fail', _ActionFail, None),
|
|
('demo_flaky', _ActionFlaky, None),
|
|
):
|
|
if not local_tx.HasHandler(name):
|
|
local_tx.register(name, action=action, compensate=compensate)
|
|
|
|
|
|
def ResetExperimentState():
|
|
"""清零重试实验状态（供 local/reset 端点调用）"""
|
|
_flaky_state['attempts'] = 0
|