diff --git a/VERSION_MATRIX.md b/VERSION_MATRIX.md new file mode 100644 index 0000000..215e373 --- /dev/null +++ b/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 全模块独立导入验证 | diff --git a/__init__.py b/__init__.py index 98c9ad0..dd37c36 100644 --- a/__init__.py +++ b/__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'） diff --git a/management/commands/gvsdsdk_check_schema.py b/management/commands/gvsdsdk_check_schema.py new file mode 100644 index 0000000..6b4b209 --- /dev/null +++ b/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 与模型定义一致' + )) diff --git a/msync/local_tx.py b/msync/local_tx.py index 956680d..3133414 100644 --- a/msync/local_tx.py +++ b/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() diff --git a/msync/migrations/0001_initial.py b/msync/migrations/0001_initial.py index 1002172..cb397da 100644 --- a/msync/migrations/0001_initial.py +++ b/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': '事务步骤', diff --git a/msync/models.py b/msync/models.py index 0932529..9388d00 100644 --- a/msync/models.py +++ b/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 diff --git a/msync/transaction.py b/msync/transaction.py index 4c4da79..8573558 100644 --- a/msync/transaction.py +++ b/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( diff --git a/pyproject.toml b/pyproject.toml index 575423b..ec33a05 100644 --- a/pyproject.toml +++ b/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"