fix(atest): 中风险项修复——重连锁协议/OGM 键名/RemoteCallDemo

- _DoReconnect 互斥锁移入内部,修复 force_deregister 路径锁协议破坏
- 订单 OGM 键名 GlobalID→ObjectGlobalID(映射不再丢失)
- RemoteCallDemo 对 requests.Response 先解析 JSON,避免 500
- 补充 Compensate 状态守卫回归用例(20/20 通过)
This commit is contained in:
TermiNexus committed 2026-08-10 23:32:10 +08:00
1 parent feee433e50
commit 1a73483db5
3 files changed
+40 -13

No files matched your search

+13 -5
View File
@@ -210,7 +210,7 @@ class OrderListView(APIView):
},
)
if ogm_result:
order.OGMMappingID = ogm_result.get('GlobalID', '')
order.OGMMappingID = ogm_result.get('ObjectGlobalID', '')
order.save(update_fields=['OGMMappingID'])
_log_sync('ogm_register', {
'object_type': 'order',
@@ -255,7 +255,7 @@ class OGMRegisterView(APIView):
_log_sync('ogm_register_manual', {
'object_type': ser.validated_data['object_type'],
'local_id': ser.validated_data['local_id'],
'global_id': result.get('global_id', ''),
'global_id': result.get('ObjectGlobalID', ''),
})
return Response(_ok(result), status=201)
@@ -304,12 +304,20 @@ class RemoteCallDemoView(APIView):
'error': '远端调用失败',
})
else:
# CallRemote 返回 requests.Response(或 dict 时兼容)
if isinstance(resp, dict):
body = resp
else:
try:
body = resp.json()
except ValueError:
body = {'code': resp.status_code, 'msg': resp.text[:200], 'data': None}
results.append({
'service': svc_name,
'domain': target_domain,
'code': resp.get('code'),
'msg': resp.get('msg'),
'data': resp.get('data'),
'code': body.get('code'),
'msg': body.get('msg'),
'data': body.get('data'),
})
return Response(_ok(results))
+4 -8
View File
@@ -59,9 +59,6 @@ class MSyncSocketClient:
def on_connect():
logger.info(f'[SIO] ← connect ns=/msync')
if self._initial_connected:
if not self._reconnect_lock.acquire(blocking=False):
logger.warning(f'[SIO] 重连已在进行中,跳过本次重连')
return
logger.info(f'[SIO] 重连 detected, 重新注册/认证')
t = threading.Thread(target=self._DoReconnect, daemon=True)
t.start()
@@ -174,11 +171,10 @@ class MSyncSocketClient:
logger.info(f'MSYNC Socket.IO 客户端已断开: {self.ServiceName}')
def _DoReconnect(self, max_retries=3):
"""重连后重新注册/认证,失败时自动重试
Args:
max_retries: 最大重试次数,默认 3 次
"""
"""重连后重新注册/认证,失败时自动重试(内部持有互斥锁)"""
if not self._reconnect_lock.acquire(blocking=False):
logger.warning('[SIO] 重连已在进行中,跳过本次重连')
return
try:
for attempt in range(1, max_retries + 1):
logger.info(f'[SIO] 重连注册/认证 第 {attempt}/{max_retries} 次')
+23
View File
@@ -211,6 +211,29 @@ def _Run():
rejected = True
check('已补偿事务 RetryStep 被拒绝', rejected)
# ─── 8. Compensate 状态守卫:PENDING/COMMITTED 拒绝补偿 ───
print('[8] Compensate 状态守卫')
dtx8 = TransactionManager.Create(
name='t8',
steps_data=[{'name': 's1', 'ParticipantService': 'svc'}],
)
rejected_pending = False
try:
TransactionManager.Compensate(dtx8.TransactionCode)
except TransactionError:
rejected_pending = True
check('PENDING 事务 Compensate 被拒绝', rejected_pending)
with mock.patch.object(TransactionManager, '_call_service', return_value={'ok': True}):
TransactionManager.Execute(dtx8.TransactionCode)
dtx8.refresh_from_db()
check('事务已提交', dtx8.TransactionStatus == 2)
rejected_committed = False
try:
TransactionManager.Compensate(dtx8.TransactionCode)
except TransactionError:
rejected_committed = True
check('COMMITTED 事务 Compensate 被拒绝', rejected_committed)
# ─── 汇总 ───
passed = sum(1 for _, ok in results if ok)
print(f'\n结果: {passed}/{len(results)} 通过')