TermiNexus
6f5537765f
- Compensate 增加状态守卫（PENDING/COMMITTED 禁止补偿） - 认证/刷新令牌过期时间戳补 UTC 时区，消除本地时区解释偏移 - on_deregister 先鉴权后清映射，未认证请求不再污染连接状态 - LocalExecute/LocalCompensate 转发 Idempotency-Key，LocalHandlers 补认证
799 lines
33 KiB
Python
799 lines
33 KiB
Python
"""模块：远程事务管理器
|
||
基于 Saga 模式的跨服务事务编排引擎，运行于主服务器（账本服务器）。
|
||
通过调用各工作服务的本地事务管理器端点来执行与补偿步骤。
|
||
|
||
架构：
|
||
主服务器 (TransactionManager)
|
||
│
|
||
├──▶ POST {service_domain}/api/msync/local/execute/
|
||
│ {StepName, payload}
|
||
│ → 工作服务的 LocalTransactionManager.execute_step()
|
||
│
|
||
└──▶ POST {service_domain}/api/msync/local/compensate/
|
||
{StepName, payload, ActionResult}
|
||
→ 工作服务的 LocalTransactionManager.compensate_step()
|
||
|
||
事务生命周期：
|
||
Pending → Running → Committed
|
||
↘ Failed → Compensating → Compensated
|
||
↘ TimedOut → Compensating → Compensated"""
|
||
|
||
import uuid
|
||
import json
|
||
import logging
|
||
import os
|
||
import datetime
|
||
import requests
|
||
from requests.adapters import HTTPAdapter
|
||
from urllib3.util.retry import Retry
|
||
from django.db import transaction as db_transaction
|
||
from gvsdsdk.msync.models import (
|
||
DistributedTransaction,
|
||
TransactionStep,
|
||
ServiceNode,
|
||
_utcnow,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
LOCAL_EXECUTE_PATH = '/api/msync/local/execute/'
|
||
LOCAL_COMPENSATE_PATH = '/api/msync/local/compensate/'
|
||
|
||
# ─── 模块级 HTTP 会话（连接池 + 重试 + TLS 验证）─────────────────
|
||
_http_session = requests.Session()
|
||
_http_session.verify = True # 显式启用 TLS 证书验证
|
||
|
||
# 支持通过环境变量指定 CA 证书包（开发/部署灵活性）
|
||
_ca_bundle = os.environ.get('REQUESTS_CA_BUNDLE') or os.environ.get('CURL_CA_BUNDLE')
|
||
if _ca_bundle:
|
||
_http_session.verify = _ca_bundle
|
||
|
||
# 配置连接池重试策略
|
||
_retry_adapter = HTTPAdapter(
|
||
max_retries=Retry(
|
||
total=3,
|
||
backoff_factor=0.5,
|
||
status_forcelist=[502, 503, 504],
|
||
allowed_methods=['POST'],
|
||
),
|
||
pool_connections=10,
|
||
pool_maxsize=10,
|
||
)
|
||
_http_session.mount('https://', _retry_adapter)
|
||
_http_session.mount('http://', _retry_adapter)
|
||
|
||
|
||
class TransactionError(Exception):
|
||
pass
|
||
|
||
|
||
class StepExecutionError(Exception):
|
||
def __init__(self, message, step_index=None, status_code=None, response_body=None):
|
||
super().__init__(message)
|
||
self.StepIndex = step_index
|
||
self.StatusCode = status_code
|
||
self.ResponseBody = response_body
|
||
|
||
|
||
class TransactionManager:
|
||
"""远程事务管理器——Saga 编排引擎（运行于主服务器）
|
||
|
||
调用各工作服务的本地事务管理器来执行步骤：
|
||
- 执行步骤 → POST {service_domain}/api/msync/local/execute/
|
||
- 补偿步骤 → POST {service_domain}/api/msync/local/compensate/
|
||
"""
|
||
|
||
@staticmethod
|
||
@db_transaction.atomic
|
||
def Create(name, steps_data, description='', transaction_type='saga',
|
||
initiator_service_name=None, reference_ogm_uuid=None,
|
||
payload=None, timeout_seconds=300):
|
||
"""创建分布式事务
|
||
|
||
Args:
|
||
name: 事务名称
|
||
steps_data: 步骤数据列表，每项为 dict:
|
||
{
|
||
'name': '步骤名称（对应本地事务管理器注册的处理器名）',
|
||
'description': '步骤描述',
|
||
'ParticipantService': '服务名称（账本中注册的服务名）',
|
||
'ActionPayload': {...},
|
||
'CompensatePayload': {...},
|
||
'MaxRetries': 3,
|
||
}
|
||
description: 事务描述
|
||
transaction_type: 事务类型 (saga/2pc)
|
||
initiator_service_name: 发起服务名称
|
||
reference_ogm_uuid: 关联 OGM UUID
|
||
payload: 事务级载荷
|
||
timeout_seconds: 超时秒数
|
||
|
||
Returns:
|
||
DistributedTransaction 实例
|
||
"""
|
||
if not steps_data:
|
||
raise TransactionError('事务至少需要一个步骤')
|
||
|
||
initiator = None
|
||
if initiator_service_name:
|
||
try:
|
||
initiator = ServiceNode.objects.get(ServiceName=initiator_service_name, ServiceStatus=1)
|
||
except ServiceNode.DoesNotExist:
|
||
raise TransactionError(f'发起服务不存在: {initiator_service_name}')
|
||
|
||
tx_code = f'DTX-{uuid.uuid4().hex[:12].upper()}'
|
||
|
||
dtx = DistributedTransaction(
|
||
TransactionUUID=uuid.uuid4().bytes,
|
||
TransactionCode=tx_code,
|
||
TransactionType=transaction_type,
|
||
TransactionName=name,
|
||
TransactionDescription=description,
|
||
TransactionStatus=DistributedTransaction.TX_STATUS_PENDING,
|
||
InitiatorServiceUUID=initiator,
|
||
ReferenceOGMUUID=reference_ogm_uuid,
|
||
TransactionPayload=payload or {},
|
||
TimeoutSeconds=timeout_seconds,
|
||
)
|
||
dtx.save()
|
||
|
||
for idx, step_data in enumerate(steps_data):
|
||
participant = None
|
||
participant_name = step_data.get('ParticipantService')
|
||
if participant_name:
|
||
try:
|
||
participant = ServiceNode.objects.get(ServiceName=participant_name, ServiceStatus=1)
|
||
except ServiceNode.DoesNotExist:
|
||
raise TransactionError(f'参与服务不存在: {participant_name}')
|
||
|
||
step = TransactionStep(
|
||
StepUUID=uuid.uuid4().bytes,
|
||
TransactionUUID=dtx,
|
||
StepIndex=idx,
|
||
StepName=step_data['name'],
|
||
StepDescription=step_data.get('description', ''),
|
||
ParticipantServiceUUID=participant,
|
||
StepStatus=TransactionStep.STEP_STATUS_PENDING,
|
||
ActionEndpoint=step_data.get('ActionEndpoint', ''),
|
||
ActionMethod='POST',
|
||
ActionPayload=step_data.get('ActionPayload', {}),
|
||
CompensateEndpoint=step_data.get('CompensateEndpoint', ''),
|
||
CompensateMethod='POST',
|
||
CompensatePayload=step_data.get('CompensatePayload', {}),
|
||
MaxRetries=step_data.get('MaxRetries', 3),
|
||
)
|
||
step.save()
|
||
|
||
logger.info(f'分布式事务创建: {tx_code} ({name}), {len(steps_data)} 步骤')
|
||
return dtx
|
||
|
||
@staticmethod
|
||
@db_transaction.atomic
|
||
def Execute(transaction_code, access_token=None):
|
||
"""执行分布式事务（Saga 正向流程）
|
||
|
||
按步骤顺序依次调用各工作服务的本地事务管理器，
|
||
任一步骤失败则触发补偿流程。
|
||
|
||
Args:
|
||
transaction_code: 事务代码
|
||
access_token: 服务间认证令牌
|
||
|
||
Returns:
|
||
DistributedTransaction 实例
|
||
"""
|
||
try:
|
||
dtx = DistributedTransaction.objects.select_for_update().get(TransactionCode=transaction_code)
|
||
except DistributedTransaction.DoesNotExist:
|
||
raise TransactionError(f'事务不存在: {transaction_code}')
|
||
|
||
if dtx.TransactionStatus not in (
|
||
DistributedTransaction.TX_STATUS_PENDING,
|
||
DistributedTransaction.TX_STATUS_RUNNING,
|
||
):
|
||
raise TransactionError(f'事务状态不允许执行: {dtx.get_TransactionStatus_display()}')
|
||
|
||
if dtx.IsTimedOut:
|
||
dtx.TransactionStatus = DistributedTransaction.TX_STATUS_TIMED_OUT
|
||
dtx.UpdateTime = _utcnow()
|
||
dtx.save(update_fields=['TransactionStatus', 'UpdateTime'])
|
||
db_transaction.on_commit(
|
||
lambda: TransactionManager.Compensate(transaction_code, access_token=access_token)
|
||
)
|
||
raise TransactionError(f'事务已超时: {transaction_code}')
|
||
|
||
dtx.TransactionStatus = DistributedTransaction.TX_STATUS_RUNNING
|
||
dtx.UpdateTime = _utcnow()
|
||
dtx.save(update_fields=['TransactionStatus', 'UpdateTime'])
|
||
|
||
steps = list(dtx.steps.all().order_by('StepIndex'))
|
||
flow_payload = {}
|
||
|
||
for step in steps:
|
||
if step.StepStatus == TransactionStep.STEP_STATUS_COMPLETED:
|
||
# 断点续跑：已完成步骤的结果仍参与后续步骤数据流
|
||
if isinstance(step.StepResult, dict):
|
||
flow_payload.update(step.StepResult)
|
||
continue
|
||
|
||
if dtx.IsTimedOut:
|
||
dtx.TransactionStatus = DistributedTransaction.TX_STATUS_TIMED_OUT
|
||
dtx.UpdateTime = _utcnow()
|
||
dtx.save(update_fields=['TransactionStatus', 'UpdateTime'])
|
||
db_transaction.on_commit(
|
||
lambda tc=transaction_code, at=access_token: TransactionManager.Compensate(tc, access_token=at)
|
||
)
|
||
raise TransactionError(f'事务执行超时: {transaction_code}')
|
||
|
||
step_succeeded = False
|
||
while not step_succeeded:
|
||
try:
|
||
result = TransactionManager._execute_step(
|
||
step,
|
||
access_token=access_token,
|
||
extra_payload=flow_payload or None,
|
||
)
|
||
step.StepStatus = TransactionStep.STEP_STATUS_COMPLETED
|
||
step.StepResult = result
|
||
step.EndTime = _utcnow()
|
||
step.save(update_fields=['StepStatus', 'StepResult', 'EndTime'])
|
||
step_succeeded = True
|
||
except StepExecutionError as e:
|
||
if step.CanRetry:
|
||
# 自动重试（未达 MaxRetries 上限），与手动 RetryStep 共享 RetryCount
|
||
step.RetryCount += 1
|
||
step.save(update_fields=['RetryCount'])
|
||
logger.warning(
|
||
f'事务步骤执行失败，自动重试 {step.RetryCount}/{step.MaxRetries}: '
|
||
f'{transaction_code} step={step.StepIndex}: {e}'
|
||
)
|
||
continue
|
||
step.StepStatus = TransactionStep.STEP_STATUS_FAILED
|
||
step.StepResult = {
|
||
'error': str(e),
|
||
'status_code': e.StatusCode,
|
||
'response': e.ResponseBody,
|
||
'retry_count': step.RetryCount,
|
||
}
|
||
step.EndTime = _utcnow()
|
||
step.save(update_fields=['StepStatus', 'StepResult', 'EndTime'])
|
||
break
|
||
|
||
if not step_succeeded:
|
||
logger.warning(f'事务步骤失败: {transaction_code} step={step.StepIndex}, 触发补偿')
|
||
db_transaction.on_commit(
|
||
lambda tc=transaction_code, at=access_token: TransactionManager.Compensate(tc, access_token=at)
|
||
)
|
||
return DistributedTransaction.objects.get(TransactionCode=transaction_code)
|
||
|
||
# 步骤数据流：本步结果合并进后续步骤载荷（仅 dict 结果）
|
||
if isinstance(step.StepResult, dict):
|
||
flow_payload.update(step.StepResult)
|
||
|
||
# 活动推进刷新空闲超时计时
|
||
dtx.UpdateTime = _utcnow()
|
||
dtx.save(update_fields=['UpdateTime'])
|
||
|
||
dtx.TransactionStatus = DistributedTransaction.TX_STATUS_COMMITTED
|
||
dtx.EndTime = _utcnow()
|
||
dtx.UpdateTime = _utcnow()
|
||
dtx.TransactionResult = TransactionManager._BuildStepSummary(dtx)
|
||
dtx.save(update_fields=['TransactionStatus', 'EndTime', 'UpdateTime', 'TransactionResult'])
|
||
|
||
logger.info(f'分布式事务提交: {transaction_code}')
|
||
return dtx
|
||
|
||
@staticmethod
|
||
@db_transaction.atomic
|
||
def Compensate(transaction_code, access_token=None):
|
||
"""补偿分布式事务（Saga 反向流程）
|
||
|
||
逆序遍历已完成步骤，调用各工作服务的本地事务管理器补偿端点。
|
||
|
||
Args:
|
||
transaction_code: 事务代码
|
||
access_token: 服务间认证令牌
|
||
"""
|
||
try:
|
||
dtx = DistributedTransaction.objects.select_for_update().get(TransactionCode=transaction_code)
|
||
except DistributedTransaction.DoesNotExist:
|
||
raise TransactionError(f'事务不存在: {transaction_code}')
|
||
|
||
if dtx.TransactionStatus == DistributedTransaction.TX_STATUS_COMPENSATED:
|
||
return dtx
|
||
|
||
# 状态守卫：仅 执行中/失败/超时/补偿中 允许补偿；
|
||
# 待执行（应走 Cancel）与已提交 一律拒绝，避免错误回滚
|
||
if dtx.TransactionStatus not in (
|
||
DistributedTransaction.TX_STATUS_RUNNING,
|
||
DistributedTransaction.TX_STATUS_FAILED,
|
||
DistributedTransaction.TX_STATUS_TIMED_OUT,
|
||
DistributedTransaction.TX_STATUS_COMPENSATING,
|
||
):
|
||
raise TransactionError(
|
||
f'事务状态不允许补偿: {dtx.get_TransactionStatus_display()}'
|
||
)
|
||
|
||
dtx.TransactionStatus = DistributedTransaction.TX_STATUS_COMPENSATING
|
||
dtx.UpdateTime = _utcnow()
|
||
dtx.save(update_fields=['TransactionStatus', 'UpdateTime'])
|
||
|
||
completed_steps = list(
|
||
dtx.steps.filter(StepStatus=TransactionStep.STEP_STATUS_COMPLETED)
|
||
.order_by('-StepIndex')
|
||
)
|
||
|
||
compensation_failed = False
|
||
for step in completed_steps:
|
||
step.StepStatus = TransactionStep.STEP_STATUS_COMPENSATING
|
||
step.save(update_fields=['StepStatus'])
|
||
|
||
try:
|
||
result = TransactionManager._compensate_step(step, access_token=access_token)
|
||
step.StepStatus = TransactionStep.STEP_STATUS_COMPENSATED
|
||
step.StepResult.update({'compensate_result': result})
|
||
step.EndTime = _utcnow()
|
||
step.save(update_fields=['StepStatus', 'StepResult', 'EndTime'])
|
||
except StepExecutionError as e:
|
||
# 补偿失败：步骤置 FAILED（而非停留在 COMPENSATING），
|
||
# 使 RetryCompensate 可再次补偿，避免事务永远补不回来
|
||
compensation_failed = True
|
||
step.StepStatus = TransactionStep.STEP_STATUS_FAILED
|
||
step.StepResult.update({
|
||
'compensate_error': str(e),
|
||
'compensate_status_code': e.StatusCode,
|
||
})
|
||
step.EndTime = _utcnow()
|
||
step.save(update_fields=['StepStatus', 'StepResult', 'EndTime'])
|
||
logger.error(f'补偿步骤失败: {transaction_code} step={step.StepIndex}: {e}')
|
||
|
||
if compensation_failed:
|
||
dtx.TransactionStatus = DistributedTransaction.TX_STATUS_FAILED
|
||
else:
|
||
dtx.TransactionStatus = DistributedTransaction.TX_STATUS_COMPENSATED
|
||
|
||
dtx.EndTime = _utcnow()
|
||
dtx.UpdateTime = _utcnow()
|
||
dtx.TransactionResult = TransactionManager._BuildStepSummary(dtx)
|
||
dtx.save(update_fields=['TransactionStatus', 'EndTime', 'UpdateTime', 'TransactionResult'])
|
||
|
||
logger.info(f'分布式事务补偿完成: {transaction_code} -> {dtx.get_TransactionStatus_display()}')
|
||
return dtx
|
||
|
||
@staticmethod
|
||
@db_transaction.atomic
|
||
def RetryStep(transaction_code, step_index, access_token=None):
|
||
"""重试失败步骤（行锁防并发）
|
||
|
||
仅允许事务处于 FAILED/RUNNING 且步骤为正向失败（无
|
||
compensate_error）时重试；手动重试不占用自动重试的
|
||
MaxRetries 配额（计数分离），重试时重建前置步骤数据流，
|
||
避免载荷漂移与幂等误命中。
|
||
"""
|
||
try:
|
||
dtx = DistributedTransaction.objects.select_for_update().get(TransactionCode=transaction_code)
|
||
except DistributedTransaction.DoesNotExist:
|
||
raise TransactionError(f'事务不存在: {transaction_code}')
|
||
|
||
if dtx.TransactionStatus not in (
|
||
DistributedTransaction.TX_STATUS_FAILED,
|
||
DistributedTransaction.TX_STATUS_RUNNING,
|
||
):
|
||
raise TransactionError(
|
||
f'事务状态不允许重试步骤: {dtx.get_TransactionStatus_display()}'
|
||
)
|
||
|
||
try:
|
||
step = dtx.steps.get(StepIndex=step_index)
|
||
except TransactionStep.DoesNotExist:
|
||
raise TransactionError(f'步骤不存在: {step_index}')
|
||
|
||
if step.StepStatus not in (
|
||
TransactionStep.STEP_STATUS_FAILED,
|
||
TransactionStep.STEP_STATUS_PENDING,
|
||
):
|
||
raise TransactionError(f'步骤状态不允许重试: {step.get_StepStatus_display()}')
|
||
|
||
if 'compensate_error' in (step.StepResult or {}):
|
||
raise TransactionError('该步骤为补偿失败，请使用 RetryCompensate 重试补偿')
|
||
|
||
# 手动重试不受自动重试 MaxRetries 配额限制（计数分离）；
|
||
# 递增 RetryCount 使幂等键随重试变化，避免命中旧失败缓存
|
||
step.RetryCount += 1
|
||
step.StepStatus = TransactionStep.STEP_STATUS_PENDING
|
||
step.save(update_fields=['RetryCount', 'StepStatus'])
|
||
|
||
# 重建步骤数据流：汇总本步骤之前所有已完成步骤的结果
|
||
flow_payload = {}
|
||
for s in dtx.steps.all().order_by('StepIndex'):
|
||
if s.StepIndex >= step.StepIndex:
|
||
break
|
||
if s.StepStatus == TransactionStep.STEP_STATUS_COMPLETED and isinstance(s.StepResult, dict):
|
||
flow_payload.update(s.StepResult)
|
||
|
||
try:
|
||
result = TransactionManager._execute_step(
|
||
step,
|
||
access_token=access_token,
|
||
extra_payload=flow_payload or None,
|
||
)
|
||
step.StepStatus = TransactionStep.STEP_STATUS_COMPLETED
|
||
step.StepResult = result
|
||
step.EndTime = _utcnow()
|
||
step.save(update_fields=['StepStatus', 'StepResult', 'EndTime'])
|
||
except StepExecutionError as e:
|
||
step.StepStatus = TransactionStep.STEP_STATUS_FAILED
|
||
step.StepResult = {
|
||
'error': str(e),
|
||
'status_code': e.StatusCode,
|
||
'response': e.ResponseBody,
|
||
'retry_count': step.RetryCount,
|
||
}
|
||
step.EndTime = _utcnow()
|
||
step.save(update_fields=['StepStatus', 'StepResult', 'EndTime'])
|
||
|
||
return step
|
||
|
||
@staticmethod
|
||
@db_transaction.atomic
|
||
def RetryCompensate(transaction_code, step_index, access_token=None):
|
||
"""重试补偿失败的步骤
|
||
|
||
仅允许对补偿失败（StepResult 含 compensate_error）的步骤重试补偿；
|
||
补偿成功后若不存在仍补偿失败的步骤，事务置 Compensated。
|
||
|
||
Args:
|
||
transaction_code: 事务代码
|
||
step_index: 步骤序号
|
||
access_token: 服务间认证令牌
|
||
|
||
Returns:
|
||
TransactionStep 实例
|
||
"""
|
||
try:
|
||
dtx = DistributedTransaction.objects.select_for_update().get(TransactionCode=transaction_code)
|
||
except DistributedTransaction.DoesNotExist:
|
||
raise TransactionError(f'事务不存在: {transaction_code}')
|
||
|
||
if dtx.TransactionStatus == DistributedTransaction.TX_STATUS_COMPENSATED:
|
||
raise TransactionError('事务已补偿完成')
|
||
|
||
try:
|
||
step = dtx.steps.get(StepIndex=step_index)
|
||
except TransactionStep.DoesNotExist:
|
||
raise TransactionError(f'步骤不存在: {step_index}')
|
||
|
||
if step.StepStatus != TransactionStep.STEP_STATUS_FAILED:
|
||
raise TransactionError(f'仅补偿失败的步骤可重试补偿: {step.get_StepStatus_display()}')
|
||
if 'compensate_error' not in (step.StepResult or {}):
|
||
raise TransactionError('该步骤为正向执行失败，请使用 RetryStep 重试执行')
|
||
|
||
if not step.CanRetry:
|
||
raise TransactionError(f'步骤已达到最大重试次数: {step.MaxRetries}')
|
||
|
||
step.RetryCount += 1
|
||
step.StepStatus = TransactionStep.STEP_STATUS_COMPENSATING
|
||
step.save(update_fields=['RetryCount', 'StepStatus'])
|
||
|
||
try:
|
||
result = TransactionManager._compensate_step(step, access_token=access_token)
|
||
step.StepStatus = TransactionStep.STEP_STATUS_COMPENSATED
|
||
step.StepResult.update({'compensate_result': result})
|
||
step.EndTime = _utcnow()
|
||
step.save(update_fields=['StepStatus', 'StepResult', 'EndTime'])
|
||
except StepExecutionError as e:
|
||
step.StepStatus = TransactionStep.STEP_STATUS_FAILED
|
||
step.StepResult.update({
|
||
'compensate_error': str(e),
|
||
'compensate_status_code': e.StatusCode,
|
||
})
|
||
step.EndTime = _utcnow()
|
||
step.save(update_fields=['StepStatus', 'StepResult', 'EndTime'])
|
||
logger.error(f'补偿重试失败: {transaction_code} step={step.StepIndex}: {e}')
|
||
|
||
dtx.TransactionStatus = DistributedTransaction.TX_STATUS_FAILED
|
||
dtx.UpdateTime = _utcnow()
|
||
dtx.save(update_fields=['TransactionStatus', 'UpdateTime'])
|
||
return step
|
||
|
||
# 补偿成功：若仍存在补偿失败的步骤则事务保持 FAILED，否则 COMPENSATED
|
||
remaining_failed = any(
|
||
s.StepStatus == TransactionStep.STEP_STATUS_FAILED
|
||
and 'compensate_error' in (s.StepResult or {})
|
||
for s in dtx.steps.all()
|
||
)
|
||
if remaining_failed:
|
||
dtx.TransactionStatus = DistributedTransaction.TX_STATUS_FAILED
|
||
else:
|
||
dtx.TransactionStatus = DistributedTransaction.TX_STATUS_COMPENSATED
|
||
dtx.UpdateTime = _utcnow()
|
||
dtx.save(update_fields=['TransactionStatus', 'UpdateTime'])
|
||
|
||
logger.info(f'补偿重试完成: {transaction_code} step={step.StepIndex} -> {dtx.get_TransactionStatus_display()}')
|
||
return step
|
||
|
||
@staticmethod
|
||
def GetDetail(transaction_code):
|
||
"""获取事务详情（含所有步骤）"""
|
||
try:
|
||
dtx = DistributedTransaction.objects.prefetch_related('steps').get(
|
||
TransactionCode=transaction_code,
|
||
)
|
||
except DistributedTransaction.DoesNotExist:
|
||
return None
|
||
|
||
steps = []
|
||
for step in dtx.steps.all().order_by('StepIndex'):
|
||
steps.append({
|
||
'Index': step.StepIndex,
|
||
'Name': step.StepName,
|
||
'Description': step.StepDescription,
|
||
'Participant': step.ParticipantServiceUUID.ServiceName if step.ParticipantServiceUUID else None,
|
||
'Status': step.get_StepStatus_display(),
|
||
'StatusCode': step.StepStatus,
|
||
'Result': step.StepResult,
|
||
'RetryCount': step.RetryCount,
|
||
'MaxRetries': step.MaxRetries,
|
||
'StartTime': step.StartTime.isoformat() if step.StartTime else None,
|
||
'EndTime': step.EndTime.isoformat() if step.EndTime else None,
|
||
})
|
||
|
||
return {
|
||
'TransactionCode': dtx.TransactionCode,
|
||
'Name': dtx.TransactionName,
|
||
'Description': dtx.TransactionDescription,
|
||
'Type': dtx.TransactionType,
|
||
'Status': dtx.get_TransactionStatus_display(),
|
||
'StatusCode': dtx.TransactionStatus,
|
||
'Initiator': dtx.InitiatorServiceUUID.ServiceName if dtx.InitiatorServiceUUID else None,
|
||
'Payload': dtx.TransactionPayload,
|
||
'Result': dtx.TransactionResult,
|
||
'TimeoutSeconds': dtx.TimeoutSeconds,
|
||
'Steps': steps,
|
||
'CreateTime': dtx.CreateTime.isoformat(),
|
||
'EndTime': dtx.EndTime.isoformat() if dtx.EndTime else None,
|
||
}
|
||
|
||
@staticmethod
|
||
def Cancel(transaction_code):
|
||
"""取消待执行的事务"""
|
||
try:
|
||
dtx = DistributedTransaction.objects.get(TransactionCode=transaction_code)
|
||
except DistributedTransaction.DoesNotExist:
|
||
raise TransactionError(f'事务不存在: {transaction_code}')
|
||
|
||
if dtx.TransactionStatus != DistributedTransaction.TX_STATUS_PENDING:
|
||
raise TransactionError(f'只有待执行状态的事务可以取消')
|
||
|
||
dtx.TransactionStatus = DistributedTransaction.TX_STATUS_FAILED
|
||
dtx.EndTime = _utcnow()
|
||
dtx.save(update_fields=['TransactionStatus', 'EndTime'])
|
||
|
||
dtx.steps.filter(StepStatus=TransactionStep.STEP_STATUS_PENDING).update(
|
||
StepStatus=TransactionStep.STEP_STATUS_SKIPPED,
|
||
EndTime=_utcnow(),
|
||
)
|
||
|
||
logger.info(f'分布式事务取消: {transaction_code}')
|
||
return dtx
|
||
|
||
@staticmethod
|
||
def _PickCompensateToken():
|
||
"""选取一条有效服务令牌用于超时补偿回调
|
||
|
||
子服务器 local 端点仅远程自省令牌有效性（不校验归属），
|
||
故任一条未过期且未撤销的令牌即可满足补偿回调认证。
|
||
"""
|
||
from gvsdsdk.msync.models import ServiceToken
|
||
return (
|
||
ServiceToken.objects
|
||
.filter(TokenStatus=1, ExpiresAt__gt=_utcnow())
|
||
.order_by('-CreateTime')
|
||
.values_list('AccessToken', flat=True)
|
||
.first()
|
||
)
|
||
|
||
@staticmethod
|
||
def CheckTimeouts(access_token=None):
|
||
"""检测所有超时事务并触发补偿
|
||
|
||
Args:
|
||
access_token: 调用方服务令牌；未提供时自动选取一条有效
|
||
令牌用于补偿回调，避免无令牌补偿被工作服务 401 拒绝。
|
||
"""
|
||
timed_out = []
|
||
running_txs = DistributedTransaction.objects.filter(
|
||
TransactionStatus__in=(
|
||
DistributedTransaction.TX_STATUS_PENDING,
|
||
DistributedTransaction.TX_STATUS_RUNNING,
|
||
),
|
||
)
|
||
|
||
if not access_token:
|
||
access_token = TransactionManager._PickCompensateToken()
|
||
|
||
for dtx in running_txs:
|
||
if dtx.IsTimedOut:
|
||
TransactionManager._mark_timed_out(dtx)
|
||
try:
|
||
TransactionManager.Compensate(
|
||
dtx.TransactionCode,
|
||
access_token=access_token,
|
||
)
|
||
except TransactionError:
|
||
pass
|
||
timed_out.append(dtx.TransactionCode)
|
||
|
||
if timed_out:
|
||
logger.warning(f'检测到 {len(timed_out)} 个超时事务: {timed_out}')
|
||
return timed_out
|
||
|
||
@staticmethod
|
||
def QueryByStatus(status, limit=50):
|
||
"""按状态查询事务"""
|
||
txs = DistributedTransaction.objects.filter(
|
||
TransactionStatus=status,
|
||
).order_by('-CreateTime')[:limit]
|
||
|
||
return [
|
||
{
|
||
'TransactionCode': tx.TransactionCode,
|
||
'Name': tx.TransactionName,
|
||
'Type': tx.TransactionType,
|
||
'Status': tx.get_TransactionStatus_display(),
|
||
'Initiator': tx.InitiatorServiceUUID.ServiceName if tx.InitiatorServiceUUID else None,
|
||
'CreateTime': tx.CreateTime.isoformat(),
|
||
'EndTime': tx.EndTime.isoformat() if tx.EndTime else None,
|
||
}
|
||
for tx in txs
|
||
]
|
||
|
||
# ─── 内部方法 ───────────────────────────────────────────
|
||
|
||
@staticmethod
|
||
def _mark_timed_out(dtx):
|
||
dtx.TransactionStatus = DistributedTransaction.TX_STATUS_TIMED_OUT
|
||
dtx.UpdateTime = _utcnow()
|
||
dtx.save(update_fields=['TransactionStatus', 'UpdateTime'])
|
||
|
||
@staticmethod
|
||
def _BuildStepSummary(dtx):
|
||
"""构建步骤汇总快照（写入事务 TransactionResult）"""
|
||
steps = []
|
||
for s in dtx.steps.all().order_by('StepIndex'):
|
||
step_result = s.StepResult or {}
|
||
steps.append({
|
||
'Index': s.StepIndex,
|
||
'Name': s.StepName,
|
||
'Status': s.get_StepStatus_display(),
|
||
'RetryCount': s.RetryCount,
|
||
'Error': step_result.get('error'),
|
||
'CompensateError': step_result.get('compensate_error'),
|
||
})
|
||
return {
|
||
'Status': dtx.get_TransactionStatus_display(),
|
||
'StatusCode': dtx.TransactionStatus,
|
||
'Steps': steps,
|
||
}
|
||
|
||
@staticmethod
|
||
def _get_service_domain(step):
|
||
"""获取步骤参与服务的域名"""
|
||
if step.ParticipantServiceUUID:
|
||
return step.ParticipantServiceUUID.ServiceDomain.rstrip('/')
|
||
return None
|
||
|
||
@staticmethod
|
||
def _execute_step(step, access_token=None, extra_payload=None):
|
||
"""执行单个步骤——调用参与服务的本地事务管理器 execute 端点"""
|
||
step.StepStatus = TransactionStep.STEP_STATUS_RUNNING
|
||
step.StartTime = _utcnow()
|
||
step.save(update_fields=['StepStatus', 'StartTime'])
|
||
|
||
service_domain = TransactionManager._get_service_domain(step)
|
||
if not service_domain:
|
||
raise StepExecutionError(
|
||
f'步骤无参与服务或服务域名未配置: step={step.StepIndex}',
|
||
step_index=step.StepIndex,
|
||
)
|
||
|
||
endpoint = f'{service_domain}{LOCAL_EXECUTE_PATH}'
|
||
payload = {
|
||
'StepName': step.StepName,
|
||
'payload': step.ActionPayload,
|
||
'TransactionCode': step.TransactionUUID.TransactionCode,
|
||
'StepIndex': step.StepIndex,
|
||
}
|
||
if extra_payload:
|
||
# 步骤间数据流：前置步骤结果合并进本步载荷（不落库，向后兼容）
|
||
payload['payload'] = {**(step.ActionPayload or {}), **extra_payload}
|
||
|
||
return TransactionManager._call_service(
|
||
endpoint=endpoint,
|
||
payload=payload,
|
||
access_token=access_token,
|
||
idempotency_key=f'{step.TransactionUUID.TransactionCode}:{step.StepIndex}:{step.RetryCount}',
|
||
)
|
||
|
||
@staticmethod
|
||
def _compensate_step(step, access_token=None):
|
||
"""补偿单个步骤——调用参与服务的本地事务管理器 compensate 端点"""
|
||
service_domain = TransactionManager._get_service_domain(step)
|
||
if not service_domain:
|
||
raise StepExecutionError(
|
||
f'步骤无参与服务或服务域名未配置: step={step.StepIndex}',
|
||
step_index=step.StepIndex,
|
||
)
|
||
|
||
endpoint = f'{service_domain}{LOCAL_COMPENSATE_PATH}'
|
||
payload = {
|
||
'StepName': step.StepName,
|
||
'payload': step.CompensatePayload,
|
||
'ActionResult': step.StepResult,
|
||
'TransactionCode': step.TransactionUUID.TransactionCode,
|
||
'StepIndex': step.StepIndex,
|
||
}
|
||
|
||
return TransactionManager._call_service(
|
||
endpoint=endpoint,
|
||
payload=payload,
|
||
access_token=access_token,
|
||
idempotency_key=f'C-{step.TransactionUUID.TransactionCode}:{step.StepIndex}:{step.RetryCount}',
|
||
)
|
||
|
||
@staticmethod
|
||
def _call_service(endpoint, payload=None, access_token=None, idempotency_key=None):
|
||
"""调用远程服务的本地事务管理器端点（使用连接池 + TLS 验证）"""
|
||
headers = {'Content-Type': 'application/json'}
|
||
if access_token:
|
||
headers['Authorization'] = f'Service {access_token}'
|
||
if idempotency_key:
|
||
# 幂等键：事务码+步骤号+重试次数，工作服务侧可按此去重（崩溃窗口内重复副作用兜底）
|
||
headers['Idempotency-Key'] = idempotency_key
|
||
|
||
try:
|
||
resp = _http_session.post(
|
||
endpoint,
|
||
json=payload or {},
|
||
headers=headers,
|
||
timeout=30,
|
||
)
|
||
response_data = resp.text
|
||
|
||
# 非 2xx：无论响应体格式（HTML 错误页/网关页等）一律视为失败，
|
||
# 避免步骤被误判为 Completed
|
||
if not resp.ok:
|
||
raise StepExecutionError(
|
||
f'本地事务管理器调用失败: HTTP {resp.status_code}',
|
||
status_code=resp.status_code,
|
||
response_body=response_data[:4096],
|
||
)
|
||
|
||
content_type = resp.headers.get('Content-Type', '')
|
||
if 'application/json' in content_type:
|
||
try:
|
||
result = resp.json()
|
||
except ValueError:
|
||
# JSONDecodeError 是 ValueError 子类：响应体声明 JSON 却无法解析时，
|
||
# 转 StepExecutionError，保证失败必进补偿链路，事务不悬挂 Running
|
||
raise StepExecutionError(
|
||
f'本地事务管理器返回非法 JSON: {response_data[:512]}',
|
||
status_code=resp.status_code,
|
||
response_body=response_data[:4096],
|
||
)
|
||
if isinstance(result, dict) and result.get('code') == 200:
|
||
return result.get('data', result)
|
||
elif isinstance(result, dict) and 'success' in result:
|
||
if not result['success']:
|
||
raise StepExecutionError(
|
||
result.get('error', '本地执行失败'),
|
||
response_body=response_data,
|
||
)
|
||
return result.get('result', result)
|
||
return result
|
||
return {'raw': response_data}
|
||
except requests.exceptions.RequestException as e:
|
||
# 统一兜底：连接失败/超时/重试耗尽等任何 requests 异常
|
||
# 一律转 StepExecutionError，确保失败必进补偿链路
|
||
raise StepExecutionError(f'本地事务管理器调用失败: {e}')
|