fix(msync): Saga 增强与已知缺陷全量修复，版本升至 1.1.0

- 自动重试：Execute 内按 CanRetry 内联重试（与手动 RetryStep 共享计数）；
  RetryStep 补 select_for_update 行锁
- 步骤数据流：前置步骤 StepResult 合并进后续步骤载荷（不落库，向后兼容）
- 空闲超时：IsTimedOut 改以 UpdateTime 起算，Execute 每步推进刷新计时
- TransactionResult：Execute/Compensate 终态写入步骤汇总快照
- 幂等键：_call_service 携带 Idempotency-Key（事务码:步骤号:重试次数），
  local_tx 侧按 key 内存去重（TTL 300s）
- 修复 RetryCompensate 成功后残留 compensate_error 导致事务误判 FAILED
- 迁移 0001 FK 去除 db_column（列名对齐模型 attname，消除 makemigrations 漂移）
- 新增 gvsdsdk_check_schema 命令：模型与数据库 schema 一致性校验
- 新增 VERSION_MATRIX.md 版本连接状态表；版本 1.0.0 -> 1.1.0

VERSION_MATRIX.md

@@ -0,0 +1,31 @@
+# GVSDSDK 版本与连接状态
+
+> 主服务器（gvsds）与子服务器（如 ATestCServer）通过 MSYNC 协议连接，
+> 双端必须安装 gvsdsdk。本表说明各版本组合的连接状态与升级建议。
+
+## 版本规则
+
+- **版本来源单一**：`gvsdsdk/__init__.py` 的 `VERSION_MAJOR/MINOR/PATCH`（pyproject.toml 同步）。
+- **启动校验**：主服务器 `gvsds/settings.py` 与子服务器 `apps/sub_server/apps.py`
+  均调用 `gvsdsdk.check_min_version(REQUIRED)`，不满足时**告警不阻断**。
+- **1.1.0 重要变更**：msync 全部实现从 `gvsds.apps.msync` 迁入 `gvsdsdk/msync`
+  （SDK 原生实现，可脱离 gvsds 独立部署）；Saga 增加自动重试/步骤数据流/空闲超时/
+  TransactionResult 汇总/幂等键；注册强制密钥认证。
+
+## 连接状态表
+
+| 主服务器 SDK | 子服务器 SDK | 连接状态 | 说明 |
+|---|---|---|---|
+| 1.0.0 | 任意 | ⛔ 不可用 | 1.0.0 主侧 msync 位于 `gvsds.apps.msync`，该路径已在 1.1.0 中删除，无法再部署 |
+| 1.1.0 | 1.1.0 | ✅ 兼容（推荐） | 版本校验通过，协议一致，全部能力可用 |
+| 1.1.0 | 1.0.0 | ⚠️ 兼容（建议升级） | 子端使用的 `gvsdsdk.msync.auth/client/local_tx` 接口 1.0.0→1.1.0 未变，可正常连接；子端启动会因 `check_min_version('1.1.0')` 不满足打告警 |
+| 1.0.0（从） | 1.1.0（主） | ⛔ 版本不满足 | 主服务器 `check_min_version('1.1.0')` 失败，功能不受保障 |
+
+> 说明：`check_min_version` 为“当前 SDK ≥ 要求版本”语义。新部署请双端统一升级到 1.1.0。
+
+## 版本历史
+
+| 版本 | 关键变化 |
+|---|---|
+| 1.0.0 | msync 为薄封装（转发 `gvsds.apps.msync`）；Saga 基础编排/补偿 |
+| 1.1.0 | msync 整体搬迁至 `gvsdsdk/msync`（原生实现，脱离 gvsds）；密钥认证强制；Saga 自动重试 / 步骤数据流 / 空闲超时 / TransactionResult / 幂等键；SDK 全模块独立导入验证 |

__init__.py

@@ -1,14 +1,13 @@
 """GVSDS SDK — GVSDS 分布式 SDK
 
 GVSDS 分布式服务间通信 SDK，作为子服务器（如 ATestCServer）和主服务器（gvsds）
-之间的统一通信桥梁。
+之间的统一通信桥梁。实现原生位于 gvsdsdk 内（非封装层），可脱离 gvsds 独立部署。
 
 设计原则：
-    1. 薄封装：直接包装 gvsds.apps.msync 原生层
-    2. 纯转发：方法签名 1:1 对应原生层
+    1. 原生实现：全部逻辑位于 gvsdsdk（含 msync 通信底座），不依赖 gvsds
+    2. 独立部署：屏蔽 gvsds 包后所有子模块可独立导入运行
     3. PascalCase：全部公开 API 使用 PascalCase
-    4. 协议兼容：完全兼容 gvsds/apps/msync 现有协议
-    5. 惰性加载：仅在访问属性时才 import 原生层（避免 Django 未初始化时报错）
+    4. 惰性加载：仅访问属性时才 import ORM 依赖模块（避免 Django 未初始化时报错）
 
 包结构：
     gvsdsdk/
@@ -17,20 +16,24 @@ GVSDS 分布式服务间通信 SDK，作为子服务器（如 ATestCServer）和
     │   ├── middleware.py    # CurrentRoleMiddleware
     │   ├── permissions.py   # RBAC/ABAC 权限
     │   └── managers.py     # RoleManager（UserManager 在 models.py）
-    ├── msync/              # MSYNC 服务间通信
+    ├── msync/              # MSYNC 服务间通信（完整实现）
+    │   ├── models.py       # 服务节点/令牌/OGM/分布式事务模型
+    │   ├── registry.py     # 服务注册发现
+    │   ├── ogm.py          # OGM 对象全局映射
+    │   ├── transaction.py  # Saga 远程事务编排
+    │   ├── local_tx.py     # 本地事务步骤处理器
+    │   ├── auth.py         # HMAC 签名 + 令牌编解码 + 服务端签发/验证
+    │   ├── sio.py          # Socket.IO 服务端
+    │   ├── views.py        # REST API 视图
+    │   └── ...
     ├── commission/         # 分佣引擎
-    │   ├── models.py       # 分佣数据模型
-    │   ├── manager.py      # CommissionEngine 编排器
-    │   └── strategies/     # 分账策略
     ├── payment/            # 支付网关
-    │   ├── models.py       # 支付数据模型
-    │   ├── manager.py      # PaymentManager 编排器
-    │   └── gateways/       # 支付渠道适配器
     ├── order/              # 订单模板与订单
-    │   └── models.py       # 订单数据模型
     ├── models.py           # SAAS 核心数据模型
     ├── model_base.py       # QModel
     ├── model_utils.py      # UUIDObj
+    ├── secrets.py          # secrets.ini 配置读取
+    ├── serializer_base.py  # PascalCase DRF 序列化器基类
     └── fluent.py           # FluentQuery ORM 查询层
 
 使用示例：
@@ -38,7 +41,7 @@ GVSDS 分布式服务间通信 SDK，作为子服务器（如 ATestCServer）和
     from gvsdsdk.auth import PermissionVerifier
     from gvsdsdk.commission import CommissionEngine
     from gvsdsdk.payment import PaymentManager
-    print(gvsdsdk.__version__)  # '1.0.0'
+    print(gvsdsdk.__version__)  # '1.1.0'
 """
 
 import os as _os
@@ -52,7 +55,7 @@ from typing import Optional, Tuple
 #: 主版本号
 VERSION_MAJOR = 1
 #: 次版本号
-VERSION_MINOR = 0
+VERSION_MINOR = 1
 #: 补丁版本号
 VERSION_PATCH = 0
 #: 预发布标签（如 'a1', 'b2', 'rc1'）

management/commands/gvsdsdk_check_schema.py

@@ -0,0 +1,80 @@
+"""gvsdsdk_check_schema — gvsdsdk/msync 模型与数据库 schema 一致性校验
+
+对比模型 db_table 的实际表结构与模型字段清单，检测：
+  - 模型有但数据库缺失的字段（需迁移）
+  - 数据库有但模型未映射的字段（模型漂移）
+
+用法：
+    python manage.py gvsdsdk_check_schema              # 校验 gvsdsdk + msync
+    python manage.py gvsdsdk_check_schema --app msync  # 仅校验 msync
+    python manage.py gvsdsdk_check_schema --check      # 仅退出码检查（CI 用）
+"""
+
+from django.apps import apps as django_apps
+from django.core.management.base import BaseCommand
+from django.db import connections, router
+
+
+class Command(BaseCommand):
+    help = '校验 gvsdsdk/msync 模型与数据库 schema 一致性（对照 db_table 字段清单）'
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--app', nargs='*', default=['gvsdsdk', 'msync'],
+            help='要校验的 app_label（默认 gvsdsdk msync）',
+        )
+        parser.add_argument(
+            '--check', action='store_true',
+            help='仅检查（存在差异时退出码非 0），不输出明细之外的汇总',
+        )
+
+    def handle(self, *args, **options):
+        app_labels = options['app']
+        check_only = options['check']
+
+        issues = []
+        checked = 0
+        for app_label in app_labels:
+            try:
+                config = django_apps.get_app_config(app_label)
+            except LookupError:
+                self.stderr.write(f'未安装的 app: {app_label}')
+                issues.append(f'{app_label}: 未安装')
+                continue
+
+            for model in config.get_models():
+                checked += 1
+                table = model._meta.db_table
+                db = router.db_for_read(model) or 'default'
+                try:
+                    cursor = connections[db].cursor()
+                    cursor.execute(f'SHOW COLUMNS FROM `{table}`')
+                    db_columns = {row[0] for row in cursor.fetchall()}
+                    cursor.close()
+                except Exception as e:
+                    issues.append(f'{app_label}.{table}: 表查询失败: {e}')
+                    continue
+
+                model_columns = {f.column for f in model._meta.fields}
+                missing = sorted(model_columns - db_columns)
+                extra = sorted(db_columns - model_columns)
+                if missing or extra:
+                    for col in missing:
+                        issues.append(f'{app_label}.{table}.{col}: 模型有但数据库缺失')
+                    for col in extra:
+                        issues.append(f'{app_label}.{table}.{col}: 数据库有但模型未映射')
+
+        if issues:
+            for issue in issues:
+                self.stderr.write(self.style.ERROR(f'[差异] {issue}'))
+            self.stderr.write(self.style.ERROR(
+                f'\n检查 {checked} 个模型，发现 {len(issues)} 处差异。'
+                '若为模型新增字段，请先生成并执行迁移；若为模型漂移，请同步 SDK 模型。'
+            ))
+            if check_only:
+                raise SystemExit(1)
+            return
+
+        self.stdout.write(self.style.SUCCESS(
+            f'OK: 检查 {checked} 个模型，schema 与模型定义一致'
+        ))

msync/local_tx.py

@@ -15,6 +15,7 @@
 """
 
 import logging
+import time
 
 logger = logging.getLogger(__name__)
 
@@ -34,8 +35,12 @@ class StepHandler:
 class LocalTransactionManager:
     """本地事务管理器——每个工作服务一个实例"""
 
+    #: 幂等缓存 TTL（秒）：主服务器崩溃窗口内重复发送同一步骤时去重
+    IDEMPOTENCY_TTL_SECONDS = 300
+
     def __init__(self):
         self._handlers = {}
+        self._idempotency_cache = {}
 
     def register(self, name, action, compensate=None):
         """注册步骤处理器"""
@@ -62,41 +67,104 @@ class LocalTransactionManager:
         """列出所有已注册的步骤处理器名称"""
         return list(self._handlers.keys())
 
-    def ExecuteStep(self, name, payload=None):
-        """执行步骤——调用本地 action 函数"""
+    # ─── 幂等去重（可选能力，收到 Idempotency-Key 时启用）────────────
+
+    def _IdempotencyGet(self, key):
+        if not key:
+            return None
+        entry = self._idempotency_cache.get(key)
+        if not entry:
+            return None
+        if time.time() - entry['ts'] > self.IDEMPOTENCY_TTL_SECONDS:
+            self._idempotency_cache.pop(key, None)
+            return None
+        return entry['result']
+
+    def _IdempotencySet(self, key, result):
+        if not key:
+            return
+        # 惰性清理：条目数超限时剔除过期项，防止缓存无限增长
+        if len(self._idempotency_cache) >= 1000:
+            now = time.time()
+            expired = [
+                k for k, v in self._idempotency_cache.items()
+                if now - v['ts'] > self.IDEMPOTENCY_TTL_SECONDS
+            ]
+            for k in expired:
+                self._idempotency_cache.pop(k, None)
+        self._idempotency_cache[key] = {'ts': time.time(), 'result': result}
+
+    def ExecuteStep(self, name, payload=None, idempotency_key=None):
+        """执行步骤——调用本地 action 函数
+
+        Args:
+            name: 步骤处理器名称
+            payload: 载荷
+            idempotency_key: 幂等键（主服务器 Idempotency-Key 头），
+                相同键在 TTL 内重复调用直接返回首次结果
+        """
+        cached = self._IdempotencyGet(idempotency_key)
+        if cached is not None:
+            logger.info(f'步骤执行幂等命中: {name} key={idempotency_key}')
+            return cached
+
         handler = self._handlers.get(name)
         if handler is None:
             msg = f'未注册的步骤处理器: {name}'
             logger.error(msg)
-            return {'success': False, 'result': None, 'error': msg}
+            result = {'success': False, 'result': None, 'error': msg}
+            self._IdempotencySet(idempotency_key, result)
+            return result
 
         try:
             result = handler.Action(payload or {})
             logger.info(f'步骤执行成功: {name}')
-            return {'success': True, 'result': result, 'error': None}
+            result = {'success': True, 'result': result, 'error': None}
         except Exception as e:
             logger.error('步骤执行失败: %s: %s', name, e, exc_info=True)
-            return {'success': False, 'result': None, 'error': str(e)}
+            result = {'success': False, 'result': None, 'error': str(e)}
+
+        self._IdempotencySet(idempotency_key, result)
+        return result
+
+    def CompensateStep(self, name, payload=None, action_result=None, idempotency_key=None):
+        """补偿步骤——调用本地 compensate 函数
+
+        Args:
+            name: 步骤处理器名称
+            payload: 载荷
+            action_result: 原始执行结果（幂等回滚依据）
+            idempotency_key: 幂等键（主服务器 Idempotency-Key 头）
+        """
+        cached = self._IdempotencyGet(idempotency_key)
+        if cached is not None:
+            logger.info(f'步骤补偿幂等命中: {name} key={idempotency_key}')
+            return cached
 
-    def CompensateStep(self, name, payload=None, action_result=None):
-        """补偿步骤——调用本地 compensate 函数"""
         handler = self._handlers.get(name)
         if handler is None:
             msg = f'未注册的步骤处理器: {name}'
             logger.error(msg)
-            return {'success': False, 'result': None, 'error': msg}
+            result = {'success': False, 'result': None, 'error': msg}
+            self._IdempotencySet(idempotency_key, result)
+            return result
 
         if handler.Compensate is None:
             logger.warning(f'步骤无补偿函数，跳过: {name}')
-            return {'success': True, 'result': None, 'error': None}
+            result = {'success': True, 'result': None, 'error': None}
+            self._IdempotencySet(idempotency_key, result)
+            return result
 
         try:
             result = handler.Compensate(payload or {}, action_result or {})
             logger.info(f'步骤补偿成功: {name}')
-            return {'success': True, 'result': result, 'error': None}
+            result = {'success': True, 'result': result, 'error': None}
         except Exception as e:
             logger.error('步骤补偿失败: %s: %s', name, e, exc_info=True)
-            return {'success': False, 'result': None, 'error': str(e)}
+            result = {'success': False, 'result': None, 'error': str(e)}
+
+        self._IdempotencySet(idempotency_key, result)
+        return result
 
 
 local_tx = LocalTransactionManager()

msync/migrations/0001_initial.py

@@ -43,7 +43,7 @@ class Migration(migrations.Migration):
                 ('RefreshExpiresAt', models.DateTimeField(db_column='RefreshExpiresAt')),
                 ('CreateTime', models.DateTimeField(db_column='CreateTime', default=datetime.datetime.utcnow)),
                 ('TokenStatus', models.SmallIntegerField(db_column='TokenStatus', default=1)),
-                ('ServiceUUID', models.ForeignKey(db_column='ServiceUUID', on_delete=django.db.models.deletion.CASCADE, related_name='tokens', to='msync.servicenode')),
+                ('ServiceUUID', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='tokens', to='msync.servicenode')),
             ],
             options={
                 'verbose_name': '服务令牌',
@@ -63,7 +63,7 @@ class Migration(migrations.Migration):
                 ('ObjectStatus', models.SmallIntegerField(db_column='ObjectStatus', default=1)),
                 ('CreateTime', models.DateTimeField(db_column='CreateTime', default=datetime.datetime.utcnow)),
                 ('UpdateTime', models.DateTimeField(blank=True, db_column='UpdateTime', null=True)),
-                ('OwnerServiceUUID', models.ForeignKey(db_column='OwnerServiceUUID', on_delete=django.db.models.deletion.CASCADE, related_name='owned_objects', to='msync.servicenode')),
+                ('OwnerServiceUUID', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='owned_objects', to='msync.servicenode')),
             ],
             options={
                 'verbose_name': '对象全局映射',
@@ -97,7 +97,7 @@ class Migration(migrations.Migration):
                 ('CreateTime', models.DateTimeField(db_column='CreateTime', default=datetime.datetime.utcnow)),
                 ('UpdateTime', models.DateTimeField(blank=True, db_column='UpdateTime', null=True)),
                 ('EndTime', models.DateTimeField(blank=True, db_column='EndTime', null=True)),
-                ('InitiatorServiceUUID', models.ForeignKey(blank=True, db_column='InitiatorServiceUUID', null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='initiated_dtx', to='msync.servicenode')),
+                ('InitiatorServiceUUID', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='initiated_dtx', to='msync.servicenode')),
             ],
             options={
                 'verbose_name': '分布式事务',
@@ -134,8 +134,8 @@ class Migration(migrations.Migration):
                 ('CreateTime', models.DateTimeField(db_column='CreateTime', default=datetime.datetime.utcnow)),
                 ('StartTime', models.DateTimeField(blank=True, db_column='StartTime', null=True)),
                 ('EndTime', models.DateTimeField(blank=True, db_column='EndTime', null=True)),
-                ('ParticipantServiceUUID', models.ForeignKey(blank=True, db_column='ParticipantServiceUUID', null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='participated_steps', to='msync.servicenode')),
-                ('TransactionUUID', models.ForeignKey(db_column='TransactionUUID', on_delete=django.db.models.deletion.CASCADE, related_name='steps', to='msync.distributedtransaction')),
+                ('ParticipantServiceUUID', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='participated_steps', to='msync.servicenode')),
+                ('TransactionUUID', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='steps', to='msync.distributedtransaction')),
             ],
             options={
                 'verbose_name': '事务步骤',

msync/models.py

@@ -233,11 +233,15 @@ class DistributedTransaction(QModel):
 
     @property
     def IsTimedOut(self):
+        # 空闲超时：以最后活动时间（UpdateTime）起算，未活动过则退回创建时间
         if self.TransactionStatus in (self.TX_STATUS_COMMITTED, self.TX_STATUS_COMPENSATED):
             return False
-        if not self.TimeoutSeconds or not self.CreateTime:
+        if not self.TimeoutSeconds:
             return False
-        elapsed = (_utcnow() - self.CreateTime.replace(tzinfo=None)).total_seconds()
+        baseline = self.UpdateTime or self.CreateTime
+        if not baseline:
+            return False
+        elapsed = (_utcnow() - baseline.replace(tzinfo=None)).total_seconds()
         return elapsed > self.TimeoutSeconds
 
 

msync/transaction.py

@@ -207,9 +207,13 @@ class TransactionManager:
         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:
@@ -221,32 +225,60 @@ class TransactionManager:
                 )
                 raise TransactionError(f'事务执行超时: {transaction_code}')
 
-            try:
-                result = TransactionManager._execute_step(step, access_token=access_token)
-                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,
-                }
-                step.EndTime = _utcnow()
-                step.save(update_fields=['StepStatus', 'StepResult', 'EndTime'])
+            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.save(update_fields=['TransactionStatus', 'EndTime', 'UpdateTime'])
+        dtx.TransactionResult = TransactionManager._BuildStepSummary(dtx)
+        dtx.save(update_fields=['TransactionStatus', 'EndTime', 'UpdateTime', 'TransactionResult'])
 
         logger.info(f'分布式事务提交: {transaction_code}')
         return dtx
@@ -310,16 +342,18 @@ class TransactionManager:
 
         dtx.EndTime = _utcnow()
         dtx.UpdateTime = _utcnow()
-        dtx.save(update_fields=['TransactionStatus', 'EndTime', 'UpdateTime'])
+        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):
-        """重试失败步骤"""
+        """重试失败步骤（行锁防并发）"""
         try:
-            dtx = DistributedTransaction.objects.get(TransactionCode=transaction_code)
+            dtx = DistributedTransaction.objects.select_for_update().get(TransactionCode=transaction_code)
         except DistributedTransaction.DoesNotExist:
             raise TransactionError(f'事务不存在: {transaction_code}')
 
@@ -424,7 +458,8 @@ class TransactionManager:
 
         # 补偿成功：若仍存在补偿失败的步骤则事务保持 FAILED，否则 COMPENSATED
         remaining_failed = any(
-            'compensate_error' in (s.StepResult or {})
+            s.StepStatus == TransactionStep.STEP_STATUS_FAILED
+            and 'compensate_error' in (s.StepResult or {})
             for s in dtx.steps.all()
         )
         if remaining_failed:
@@ -554,6 +589,26 @@ class TransactionManager:
         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):
         """获取步骤参与服务的域名"""
@@ -562,7 +617,7 @@ class TransactionManager:
         return None
 
     @staticmethod
-    def _execute_step(step, access_token=None):
+    def _execute_step(step, access_token=None, extra_payload=None):
         """执行单个步骤——调用参与服务的本地事务管理器 execute 端点"""
         step.StepStatus = TransactionStep.STEP_STATUS_RUNNING
         step.StartTime = _utcnow()
@@ -582,11 +637,15 @@ class TransactionManager:
             '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
@@ -612,14 +671,18 @@ class TransactionManager:
             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):
+    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(

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "gvsdsdk"
-version = "1.0.0"
+version = "1.1.0"
 description = "GVSDS SDK — GVSDS 平台主从服务器间共享通信桥梁（gvsds 主服务器与子服务器共用）"
 readme = "README.md"
 requires-python = ">=3.11"