From d46e3a22c5e58a86b4445f157d52849e6fe91ca6 Mon Sep 17 00:00:00 2001 From: TermiNexus Date: Mon, 10 Aug 2026 22:39:08 +0800 Subject: [PATCH] =?UTF-8?q?fix(msync):=20=E7=89=88=E6=9C=AC=E5=8D=87?= =?UTF-8?q?=E8=87=B3=201.1.0=EF=BC=8CSaga=20=E7=BC=BA=E9=99=B7=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=20+=20ATestCServer=20=E6=9C=AC=E5=9C=B0=E4=BA=8B?= =?UTF-8?q?=E5=8A=A1=E7=AB=AF=E7=82=B9=20+=20=E6=96=87=E6=A1=A3=E4=BF=AE?= =?UTF-8?q?=E6=AD=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - gvsds/settings.py 版本要求升至 1.1.0 - 新增 Saga 状态机回归测试（sqlite 内存库，12 项断言，独立运行） - ATestCServer：新增 local/execute、local/compensate、local/handlers 端点 （主服务器 Saga 驱动子服务器步骤），主服务器令牌经 introspect 远程校验 - SAAS_ARCHITECTURE.md：msync 双包描述更新、Role/Permission 双份收敛标注、 ABAC 定位更正为属性标识符补充层 --- ATestCServer/apps/sub_server/apps.py | 2 +- ATestCServer/apps/sub_server/service_auth.py | 81 ++++++++ ATestCServer/apps/sub_server/urls.py | 3 + ATestCServer/apps/sub_server/views.py | 72 ++++++- SAAS_ARCHITECTURE.md | 13 +- gvsds/settings.py | 2 +- gvsds/tests/test_msync_saga.py | 178 ++++++++++++++++++ gvsdsdk/VERSION_MATRIX.md | 31 +++ gvsdsdk/__init__.py | 33 ++-- .../commands/gvsdsdk_check_schema.py | 80 ++++++++ gvsdsdk/msync/local_tx.py | 90 +++++++-- gvsdsdk/msync/migrations/0001_initial.py | 10 +- gvsdsdk/msync/models.py | 8 +- gvsdsdk/msync/transaction.py | 107 ++++++++--- gvsdsdk/pyproject.toml | 2 +- 15 files changed, 646 insertions(+), 66 deletions(-) create mode 100644 ATestCServer/apps/sub_server/service_auth.py create mode 100644 gvsds/tests/test_msync_saga.py create mode 100644 gvsdsdk/VERSION_MATRIX.md create mode 100644 gvsdsdk/management/commands/gvsdsdk_check_schema.py diff --git a/ATestCServer/apps/sub_server/apps.py b/ATestCServer/apps/sub_server/apps.py index 9981a16d..9f1ccf49 100644 --- a/ATestCServer/apps/sub_server/apps.py +++ b/ATestCServer/apps/sub_server/apps.py @@ -9,7 +9,7 @@ from django.conf import settings logger = logging.getLogger(__name__) # 与主服务器约定的 SDK 最低版本——版本不一致时启动警告 -REQUIRED_GVSDSDK_VERSION = '1.0.0' +REQUIRED_GVSDSDK_VERSION = '1.1.0' class SubServerConfig(AppConfig): diff --git a/ATestCServer/apps/sub_server/service_auth.py b/ATestCServer/apps/sub_server/service_auth.py new file mode 100644 index 00000000..c9c7c307 --- /dev/null +++ b/ATestCServer/apps/sub_server/service_auth.py @@ -0,0 +1,81 @@ +"""子服务器服务认证 + +主服务器调用子服务器 local 端点时携带 `Authorization: Service `。 +子服务器无法本地验证令牌（无主库 ServiceToken 表 / MASTER_SECRET）， +因此通过主服务器 `/api/msync/auth/introspect/` 远程自省校验。 +""" + +import json +import logging +import urllib.request +import urllib.error + +from django.conf import settings +from rest_framework.authentication import BaseAuthentication +from rest_framework.exceptions import AuthenticationFailed + +logger = logging.getLogger(__name__) + +SERVICE_AUTH_HEADER = 'Authorization' +SERVICE_AUTH_PREFIX = 'Service' +INTROSPECT_PATH = '/api/msync/auth/introspect/' + + +class ServicePrincipal: + """服务主体——标识调用方服务""" + + def __init__(self, service_name): + self.service_name = service_name + self.service_scope = [] + + @property + def is_authenticated(self): + return True + + def __str__(self): + return f'Service:{self.service_name}' + + +class SubServerServiceAuthentication(BaseAuthentication): + """DRF 认证类——校验主服务器 Service 令牌（远程自省）""" + + def authenticate(self, request): + auth_header = request.headers.get(SERVICE_AUTH_HEADER, '') + if not auth_header.startswith(f'{SERVICE_AUTH_PREFIX} '): + return None + token = auth_header[len(SERVICE_AUTH_PREFIX) + 1:].strip() + if not token: + return None + + payload = self._Introspect(token) + if not payload: + raise AuthenticationFailed('服务令牌验证失败') + + return (ServicePrincipal(payload.get('sub', 'unknown')), token) + + def authenticate_header(self, request): + return 'Service' + + @staticmethod + def _Introspect(access_token): + """调用主服务器令牌自省端点""" + master_url = str(getattr(settings, 'MSYNC_MASTER_URL', '')).rstrip('/') + if not master_url: + return None + body = json.dumps({'AccessToken': access_token}).encode('utf-8') + req = urllib.request.Request( + f'{master_url}{INTROSPECT_PATH}', + data=body, + headers={'Content-Type': 'application/json'}, + method='POST', + ) + try: + with urllib.request.urlopen(req, timeout=10) as resp: + data = json.loads(resp.read().decode('utf-8')) + except (urllib.error.URLError, urllib.error.HTTPError, ValueError, OSError) as e: + logger.error(f'令牌自省失败: {e}') + return None + + if not data.get('Valid'): + return None + return data.get('Payload') or {} diff --git a/ATestCServer/apps/sub_server/urls.py b/ATestCServer/apps/sub_server/urls.py index c90956e7..d3a8bb62 100644 --- a/ATestCServer/apps/sub_server/urls.py +++ b/ATestCServer/apps/sub_server/urls.py @@ -15,4 +15,7 @@ urlpatterns = [ path('ogm/resolve//', views.OGMResolveView.as_view(), name='ogm_resolve'), path('remote/call/', views.RemoteCallDemoView.as_view(), name='remote_call'), path('sync/logs/', views.SyncLogView.as_view(), name='sync_logs'), + path('local/execute/', views.LocalExecuteView.as_view(), name='local_execute'), + path('local/compensate/', views.LocalCompensateView.as_view(), name='local_compensate'), + path('local/handlers/', views.LocalHandlersView.as_view(), name='local_handlers'), ] \ No newline at end of file diff --git a/ATestCServer/apps/sub_server/views.py b/ATestCServer/apps/sub_server/views.py index 2d7de32f..ab4daf48 100644 --- a/ATestCServer/apps/sub_server/views.py +++ b/ATestCServer/apps/sub_server/views.py @@ -6,7 +6,8 @@ from django.conf import settings from django.shortcuts import render from rest_framework.views import APIView from rest_framework.response import Response -from rest_framework.permissions import AllowAny +from rest_framework.permissions import AllowAny, IsAuthenticated +from .service_auth import SubServerServiceAuthentication from .models import BusinessUser, BusinessOrder, SyncLog from .serializers import ( BusinessUserSerializer, @@ -329,4 +330,71 @@ def StatusPageView(request): 'service_name': settings.MSYNC_SERVICE_NAME, 'service_domain': settings.MSYNC_SERVICE_DOMAIN, 'master_url': settings.MSYNC_MASTER_URL, - }) \ No newline at end of file + }) + + +class LocalExecuteView(APIView): + """本地事务执行端点——由主服务器 Saga 编排引擎调用（需主服务器 Service 令牌）""" + + authentication_classes = [SubServerServiceAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request): + from gvsdsdk.msync.local_tx import local_tx + + step_name = request.data.get('StepName', '') + payload = request.data.get('payload') or {} + idempotency_key = request.headers.get('Idempotency-Key') + + if not step_name: + return Response(_err('StepName 不能为空', 400), status=400) + + result = local_tx.ExecuteStep( + name=step_name, + payload=payload, + idempotency_key=idempotency_key, + ) + + if result['success']: + return Response(_ok(result['result'])) + return Response(_err(result['error'], 500), status=500) + + +class LocalCompensateView(APIView): + """本地事务补偿端点——由主服务器 Saga 编排引擎调用（需主服务器 Service 令牌）""" + + authentication_classes = [SubServerServiceAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request): + from gvsdsdk.msync.local_tx import local_tx + + step_name = request.data.get('StepName', '') + payload = request.data.get('payload') or {} + action_result = request.data.get('ActionResult') or {} + idempotency_key = request.headers.get('Idempotency-Key') + + if not step_name: + return Response(_err('StepName 不能为空', 400), status=400) + + result = local_tx.CompensateStep( + name=step_name, + payload=payload, + action_result=action_result, + idempotency_key=idempotency_key, + ) + + if result['success']: + return Response(_ok(result['result'])) + return Response(_err(result['error'], 500), status=500) + + +class LocalHandlersView(APIView): + """列出已注册的本地事务步骤处理器""" + + authentication_classes = [SubServerServiceAuthentication] + permission_classes = [IsAuthenticated] + + def get(self, request): + from gvsdsdk.msync.local_tx import local_tx + return Response(_ok(local_tx.ListHandlers())) \ No newline at end of file diff --git a/SAAS_ARCHITECTURE.md b/SAAS_ARCHITECTURE.md index 898336df..9fe3a0b9 100644 --- a/SAAS_ARCHITECTURE.md +++ b/SAAS_ARCHITECTURE.md @@ -59,12 +59,13 @@ - **服务层**：从薄封装演进为**通用 SAAS 服务层**——`RoleManager / AuthManager / AdminUserManager / TenantManager / CompanyManager / DepartmentManager / MessageManager / MenuManager / WorkOrderManager / InventoryManager / AssetManager / AuditManager / OrderManager / StatsManager / UserService` 等，gvsds 视图层通过它们访问数据（裸 ORM 已基本消除） - **惰性加载（PEP 562）**：130+ 符号按需 import -- **双类导出**：自有实现（无 Django 依赖，子服务器可用）+ 服务端转发（依赖 Django ORM） +- **双类导出**：自有实现（无 Django 依赖，子服务器可用）+ 服务端实现（依赖 Django ORM） +- **msync 原生实现（1.1.0）**：通信底座已从 `gvsds.apps.msync` 整体迁入 `gvsdsdk/msync`（models/registry/ogm/transaction/auth + Django 集成层），可脱离 gvsds 独立部署（屏蔽 gvsds 后全模块独立导入验证通过） - **引擎**：`CommissionEngine`（分佣）、`PaymentManager`（支付）、`OrderManager`（订单状态机）、`StatsManager`（统计）、`TransactionManager`（分布式事务） ### 3.3 主从通信机制 -- **版本校验**：启动时 `gvsdsdk.check_min_version('1.0.0')`，主从必须一致 +- **版本校验**：启动时 `gvsdsdk.check_min_version('1.1.0')`，主从必须一致（版本兼容矩阵见 `gvsdsdk/VERSION_MATRIX.md`） - **Socket.IO 协议**：20+ 事件，全程 PascalCase 字段 - **支付网关自动注册**：`gvsdsdk/apps.py` 的 `ready()` 钩子 @@ -213,9 +214,9 @@ HTTP Request | 优先级 | 问题 | 影响 | |---|---|---| -| 高 | Role/Permission 模型在 gvsds 与 gvsdsdk 双份定义 | 字段漂移风险（服务层化已缓解视图层裸访问，模型层残留） | -| 高 | 部分表（User/Dept/Workgroup/Shop）无 TenantUUID | 跨租户查询需 JOIN，删除租户困难 | -| 中 | ABAC 实现本质是 RBAC（字符串集合） | 名实不符，扩展受限 | +| ~~高~~ | ~~Role/Permission 模型双份定义~~：✅ 已收敛（2026-08-10）——gvsds 各 app 模型统一 `from gvsdsdk.models import`，SDK 为唯一模型源 | — | +| 高 | 部分表（User/Dept/Workgroup/Shop）无 TenantUUID | 跨租户查询需 JOIN，删除租户困难（待迁移脚本补列回填） | +| 中 | ABAC 实现本质是 RBAC（字符串集合） | 已文档更正为"属性标识符补充层"，真 ABAC 重构列入中期规划 | | 中 | 数据权限硬编码 `if table_name == 'X'` | 违反开闭原则 | | 中 | 缺乏软删除机制 | 误删不可恢复 | | ~~中~~ | ~~stats 双轨~~：✅ 已统一（2026-08-07）——`StatEngine`（gvsds/apps/stats/manager.py）已删除，编排方法全部并入 gvsdsdk `StatsManager`，视图/Celery 任务统一走 SDK | — | @@ -227,7 +228,7 @@ HTTP Request ### 7.1 短期（1-3 个月） -1. ~~统一 Role/Permission 模型~~：✅ 服务层化已让视图层统一经 gvsdsdk Manager 访问（模型层双份仍待收敛） +1. ~~统一 Role/Permission 模型~~：✅ 已收敛（2026-08-10）——视图层统一经 gvsdsdk Manager 访问，模型层 gvsds 统一从 gvsdsdk import，双份定义消除 2. **补全 TenantUUID 字段**：所有租户语义表直接打 `TenantUUID`，简化跨租户查询与租户删除 3. **数据权限策略化**：用策略模式替代 `if table_name == 'X'`，每个模型注册自己的 `ScopeFilter` 4. ~~更新根 README.md~~：✅ 已反映当前 Django 架构 diff --git a/gvsds/settings.py b/gvsds/settings.py index b9ebb33c..6fbbd532 100644 --- a/gvsds/settings.py +++ b/gvsds/settings.py @@ -14,7 +14,7 @@ if _GVSDSDK not in _sys.path: _sys.path.insert(0, _GVSDSDK) # 启动时校验 gvsdsdk 版本——主从必须一致 -_REQUIRED_GVSDSDK = '1.0.0' +_REQUIRED_GVSDSDK = '1.1.0' try: import gvsdsdk _sdk_info = gvsdsdk.get_version_info() diff --git a/gvsds/tests/test_msync_saga.py b/gvsds/tests/test_msync_saga.py new file mode 100644 index 00000000..aa5514de --- /dev/null +++ b/gvsds/tests/test_msync_saga.py @@ -0,0 +1,178 @@ +"""MSYNC Saga 状态机回归测试（独立运行，不依赖主库） + +用法： + python gvsds/tests/test_msync_saga.py + +覆盖： + - 正向全部成功 → Committed，且写入 TransactionResult 汇总 + - 步骤失败 → 逆序补偿已完成步骤 → Compensated + - 自动重试：步骤失败后按 CanRetry 内联重试成功 → Committed + - 补偿失败 → 事务 FAILED，RetryCompensate 成功后 → COMPENSATED + - 本地事务处理器幂等去重（同 Idempotency-Key 返回首次结果） + +说明：远程调用通过 mock 注入，不发起真实 HTTP；模型建于 sqlite 内存库。 +""" + +import os +import sys + +# 将项目根加入 sys.path（脚本可能从任意 cwd 运行） +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) + +import django +from django.conf import settings + +_TEST_SETTINGS = { + 'DEBUG': True, + 'SECRET_KEY': 'msync-test-key', + 'USE_TZ': False, + 'DATABASES': { + 'default': { + 'ENGINE': 'django.db.backends.sqlite3', + 'NAME': ':memory:', + }, + }, + 'INSTALLED_APPS': [ + 'django.contrib.contenttypes', + 'django.contrib.auth', + 'gvsdsdk', + 'gvsdsdk.msync', + ], +} + + +def _Setup(): + settings.configure(**_TEST_SETTINGS) + django.setup() + from django.core.management import call_command + call_command('migrate', run_syncdb=True, verbosity=0, interactive=False) + + +def _Run(): + _Setup() + + from unittest import mock + import uuid + from gvsdsdk.msync.transaction import TransactionManager, StepExecutionError + from gvsdsdk.msync.local_tx import local_tx + from gvsdsdk.msync.models import ServiceNode + + # 注册参与服务节点（Saga 步骤的 ParticipantService） + ServiceNode.objects.create( + ServiceUUID=uuid.uuid4().bytes, + ServiceName='svc', + ServiceDomain='http://svc.local', + ServiceSecretHash='x', + ServiceStatus=1, + ) + + results = [] + + def check(name, cond): + results.append((name, cond)) + print(f' {"PASS" if cond else "FAIL"} {name}') + + # ─── 1. 正向全部成功 → Committed ─── + print('[1] 正向全部成功') + dtx = TransactionManager.Create( + name='t1', + steps_data=[ + {'name': 's1', 'ParticipantService': 'svc', 'ActionPayload': {'a': 1}}, + {'name': 's2', 'ParticipantService': 'svc', 'ActionPayload': {'b': 2}}, + ], + ) + with mock.patch.object(TransactionManager, '_call_service', side_effect=[ + {'user_uuid': 'u1'}, # s1 结果 → 数据流注入 s2 + {'ok': True}, + ]) as call: + TransactionManager.Execute(dtx.TransactionCode) + dtx.refresh_from_db() + check('状态为 Committed', dtx.TransactionStatus == 2) + check('TransactionResult 已写入', bool(dtx.TransactionResult and dtx.TransactionResult.get('Steps'))) + # 数据流：s2 的 payload 应包含 s1 结果（user_uuid） + s2_payload = call.call_args_list[1].kwargs['payload']['payload'] + check('步骤数据流（s1 结果注入 s2）', s2_payload.get('user_uuid') == 'u1') + + # ─── 2. 步骤失败 → 补偿已完成步骤 → Compensated ─── + print('[2] 步骤失败触发补偿') + dtx = TransactionManager.Create( + name='t2', + steps_data=[ + {'name': 's1', 'ParticipantService': 'svc'}, + {'name': 's2', 'ParticipantService': 'svc', 'MaxRetries': 0}, + {'name': 's3', 'ParticipantService': 'svc'}, + ], + ) + with mock.patch.object(TransactionManager, '_call_service', side_effect=[ + {'ok': 1}, # s1 成功 + StepExecutionError('s2 失败'), # s2 失败（自动重试耗尽） + {'ok': 'compensated'}, # 补偿 s1 + ]): + TransactionManager.Execute(dtx.TransactionCode) + dtx.refresh_from_db() + check('状态为 Compensated', dtx.TransactionStatus == 4) + steps = {s.StepIndex: s for s in dtx.steps.all()} + check('s1 已补偿', steps[0].StepStatus == 4) + check('s2 正向失败', steps[1].StepStatus == 5) + + # ─── 3. 自动重试：首次失败后成功 → Committed ─── + print('[3] 自动重试') + dtx = TransactionManager.Create( + name='t3', + steps_data=[ + {'name': 's1', 'ParticipantService': 'svc', 'MaxRetries': 2}, + ], + ) + with mock.patch.object(TransactionManager, '_call_service', side_effect=[ + StepExecutionError('临时失败'), + {'ok': True}, + ]): + TransactionManager.Execute(dtx.TransactionCode) + dtx.refresh_from_db() + step = dtx.steps.first() + check('状态为 Committed', dtx.TransactionStatus == 2) + check('RetryCount=1', step.RetryCount == 1) + + # ─── 4. 补偿失败 → Failed；RetryCompensate 成功后 → Compensated ─── + print('[4] 补偿失败可重试') + dtx = TransactionManager.Create( + name='t4', + steps_data=[ + {'name': 's1', 'ParticipantService': 'svc'}, + {'name': 's2', 'ParticipantService': 'svc', 'MaxRetries': 0}, + ], + ) + with mock.patch.object(TransactionManager, '_call_service', side_effect=[ + {'ok': 1}, + StepExecutionError('s2 失败'), + StepExecutionError('补偿 s1 失败'), + ]): + TransactionManager.Execute(dtx.TransactionCode) + dtx.refresh_from_db() + check('补偿失败后事务 Failed', dtx.TransactionStatus == 5) + # RetryCompensate：先成功补偿 s1 + with mock.patch.object(TransactionManager, '_call_service', return_value={'ok': True}): + TransactionManager.RetryCompensate(dtx.TransactionCode, 0) + dtx.refresh_from_db() + check('RetryCompensate 后 Compensated', dtx.TransactionStatus == 4) + + # ─── 5. 本地事务幂等去重 ─── + print('[5] local_tx 幂等') + r1 = local_tx.ExecuteStep('no_such', {}, 'K:1:0') + r2 = local_tx.ExecuteStep('no_such', {}, 'K:1:0') + check('同键幂等命中', r1 is r2) + r3 = local_tx.ExecuteStep('no_such', {}, 'K:1:1') + check('不同键不命中', r1 is not r3) + + # ─── 汇总 ─── + passed = sum(1 for _, ok in results if ok) + print(f'\n结果: {passed}/{len(results)} 通过') + if passed != len(results): + for name, ok in results: + if not ok: + print(f' 未通过: {name}') + sys.exit(1) + + +if __name__ == '__main__': + _Run() diff --git a/gvsdsdk/VERSION_MATRIX.md b/gvsdsdk/VERSION_MATRIX.md new file mode 100644 index 00000000..215e373d --- /dev/null +++ b/gvsdsdk/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/gvsdsdk/__init__.py b/gvsdsdk/__init__.py index 98c9ad04..dd37c366 100644 --- a/gvsdsdk/__init__.py +++ b/gvsdsdk/__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/gvsdsdk/management/commands/gvsdsdk_check_schema.py b/gvsdsdk/management/commands/gvsdsdk_check_schema.py new file mode 100644 index 00000000..6b4b2098 --- /dev/null +++ b/gvsdsdk/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/gvsdsdk/msync/local_tx.py b/gvsdsdk/msync/local_tx.py index 956680da..3133414e 100644 --- a/gvsdsdk/msync/local_tx.py +++ b/gvsdsdk/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/gvsdsdk/msync/migrations/0001_initial.py b/gvsdsdk/msync/migrations/0001_initial.py index 10021727..cb397daf 100644 --- a/gvsdsdk/msync/migrations/0001_initial.py +++ b/gvsdsdk/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/gvsdsdk/msync/models.py b/gvsdsdk/msync/models.py index 0932529e..9388d005 100644 --- a/gvsdsdk/msync/models.py +++ b/gvsdsdk/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/gvsdsdk/msync/transaction.py b/gvsdsdk/msync/transaction.py index 4c4da79f..85735581 100644 --- a/gvsdsdk/msync/transaction.py +++ b/gvsdsdk/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/gvsdsdk/pyproject.toml b/gvsdsdk/pyproject.toml index 575423bb..ec33a055 100644 --- a/gvsdsdk/pyproject.toml +++ b/gvsdsdk/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"