diff --git a/ATestCServer/ATestCServer/urls.py b/ATestCServer/ATestCServer/urls.py index 3f0a559d..79a124b6 100644 --- a/ATestCServer/ATestCServer/urls.py +++ b/ATestCServer/ATestCServer/urls.py @@ -3,6 +3,7 @@ from django.urls import path, include from django.shortcuts import redirect from apps.sub_server.views import HealthView +from apps.sub_server import views def root_redirect(request): @@ -13,4 +14,8 @@ urlpatterns = [ path('', root_redirect), path('health/', HealthView.as_view(), name='health'), path('api/sub/', include('apps.sub_server.urls')), + # SDK 契约路径：主服务器 Saga 编排引擎按 /api/msync/local/* 回调 + path('api/msync/local/execute/', views.LocalExecuteView.as_view(), name='msync_local_execute'), + path('api/msync/local/compensate/', views.LocalCompensateView.as_view(), name='msync_local_compensate'), + path('api/msync/local/reset/', views.LocalResetView.as_view(), name='msync_local_reset'), ] \ No newline at end of file diff --git a/ATestCServer/apps/sub_server/apps.py b/ATestCServer/apps/sub_server/apps.py index 9f1ccf49..6e3fdab0 100644 --- a/ATestCServer/apps/sub_server/apps.py +++ b/ATestCServer/apps/sub_server/apps.py @@ -21,9 +21,19 @@ class SubServerConfig(AppConfig): def ready(self): # 1) 校验 gvsdsdk 版本与主服务器一致 self._check_sdk_version() - # 2) 启动 MSYNC 桥接 + # 2) 注册实验本地事务处理器（供主服务器 Saga 编排回调） + self._register_local_handlers() + # 3) 启动 MSYNC 桥接 self._connect_msync() + def _register_local_handlers(self): + try: + from .local_handlers import RegisterExperimentHandlers + RegisterExperimentHandlers() + logger.info('实验本地事务处理器已注册: demo_a/demo_b/demo_fail/demo_flaky') + except Exception as e: + logger.error(f'实验本地事务处理器注册失败: {e}') + def _check_sdk_version(self): try: import gvsdsdk diff --git a/ATestCServer/apps/sub_server/local_handlers.py b/ATestCServer/apps/sub_server/local_handlers.py new file mode 100644 index 00000000..a97818fd --- /dev/null +++ b/ATestCServer/apps/sub_server/local_handlers.py @@ -0,0 +1,53 @@ +"""实验用本地事务处理器 + +随子服务器进程（runserver）启动即注册，供主服务器 Saga 编排回调。 +msync_experiment 管理命令与 HTTP 服务共用同一进程内处理器。 +""" + +import uuid + +from gvsdsdk.msync.local_tx import local_tx + +# demo_flaky 重试实验状态（进程内计数器，可通过 local/reset 端点清零） +_flaky_state = {'attempts': 0} + + +def _ActionA(payload): + return {'demo_a_uuid': str(uuid.uuid4()), 'received': payload} + + +def _ActionB(payload): + # 返回收到的完整 payload，供主服务器验证步骤间数据流 + return {'received_payload': payload} + + +def _CompensateA(payload, action_result): + return {'compensated': True, 'action_result': action_result} + + +def _ActionFail(payload): + raise RuntimeError('确定性失败（实验）') + + +def _ActionFlaky(payload): + _flaky_state['attempts'] += 1 + if _flaky_state['attempts'] == 1: + raise RuntimeError('首次失败（实验：验证自动重试）') + return {'retried': True, 'attempts': _flaky_state['attempts']} + + +def RegisterExperimentHandlers(): + """注册全部实验处理器（幂等：已存在则跳过）""" + for name, action, compensate in ( + ('demo_a', _ActionA, _CompensateA), + ('demo_b', _ActionB, None), + ('demo_fail', _ActionFail, None), + ('demo_flaky', _ActionFlaky, None), + ): + if not local_tx.HasHandler(name): + local_tx.register(name, action=action, compensate=compensate) + + +def ResetExperimentState(): + """清零重试实验状态（供 local/reset 端点调用）""" + _flaky_state['attempts'] = 0 diff --git a/ATestCServer/apps/sub_server/management/__init__.py b/ATestCServer/apps/sub_server/management/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ATestCServer/apps/sub_server/management/commands/__init__.py b/ATestCServer/apps/sub_server/management/commands/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ATestCServer/apps/sub_server/management/commands/msync_experiment.py b/ATestCServer/apps/sub_server/management/commands/msync_experiment.py new file mode 100644 index 00000000..cc0d80bc --- /dev/null +++ b/ATestCServer/apps/sub_server/management/commands/msync_experiment.py @@ -0,0 +1,202 @@ +"""msync_experiment — MSYNC 主从链路自动化实验与验证 + +用法: + python manage.py msync_experiment + +前置条件: + - 主服务器在线（SIO + HTTP），且本服务已通过 MSYNC 桥接注册成功 + - 本服务 local 端点可被主服务器访问（MSYNC_SERVICE_DOMAIN） + +自动执行并输出 PASS/FAIL 汇报表: + 1. 环境状态: SDK 版本 / SIO 连接 / 认证 / 注册 + 2. 正向 Saga: 多步骤成功 → Committed，且验证步骤间数据流 + 3. 失败补偿: 步骤失败 → 逆序补偿 → Compensated + 4. 自动重试: 步骤首次失败 → 主服务器按 CanRetry 自动重试成功 + 5. 幂等去重: local_tx 相同 Idempotency-Key 命中缓存 +""" + +import logging +import time + +from django.conf import settings +from django.core.management.base import BaseCommand + +logger = logging.getLogger(__name__) + + +def _get_msync(): + return getattr(settings, 'MSYNC_CLIENT', None) + + +class Command(BaseCommand): + help = 'MSYNC 主从链路自动化实验与验证' + + def add_arguments(self, parser): + parser.add_argument( + '--wait-seconds', type=int, default=10, + help='等待桥接就绪的秒数（默认 10）', + ) + + def handle(self, *args, **options): + self.results = [] + wait_seconds = options['wait_seconds'] + + self.out('=' * 60) + self.out('MSYNC 主从链路自动化实验') + self.out('=' * 60) + + msync = _get_msync() + if msync is None: + self.fail('MSYNC 未初始化', 'settings.MSYNC_CLIENT 不存在，桥接线程未启动') + return + + # 1. 等待桥接就绪（连接/注册/认证） + self.record('等待桥接就绪', self._wait_ready(msync, wait_seconds)) + self._report_status(msync) + + # 2. 注册本地实验步骤处理器 + self.record('注册本地步骤处理器', self._register_handlers(), 'demo_a/demo_b/demo_fail') + + # 3. 正向 Saga + 步骤数据流 + self.record('正向 Saga（数据流）', self._experiment_forward(msync)) + + # 4. 失败补偿 + self.record('失败补偿', self._experiment_compensate(msync)) + + # 5. 自动重试 + self.record('自动重试', self._experiment_retry(msync)) + + # 6. 幂等去重 + self.record('幂等去重', self._experiment_idempotency()) + + self._summary() + + # ─── 工具 ────────────────────────────────────────────── + + def out(self, msg): + self.stdout.write(msg) + + def record(self, name, ok, detail=''): + self.results.append((name, bool(ok), detail or '')) + mark = 'PASS' if ok else 'FAIL' + self.stdout.write(f' [{mark}] {name}' + (f' ({detail})' if detail else '')) + + def fail(self, name, detail=''): + self.record(name, False, detail) + + def _wait_ready(self, msync, seconds): + for _ in range(seconds * 2): + if msync.IsConnected and msync.IsAuthenticated and msync.IsRegistered: + return True + time.sleep(0.5) + return msync.IsConnected and msync.IsAuthenticated and msync.IsRegistered + + def _report_status(self, msync): + import gvsdsdk + info = gvsdsdk.get_version_info() + self.out(f'\n SDK 版本 : {info["Version"]} @ {info["Location"]}') + self.out(f' SIO 连接 : {msync.IsConnected}') + self.out(f' 密钥认证 : {msync.IsAuthenticated}') + self.out(f' 服务注册 : {msync.IsRegistered}') + self.out(f' 服务名称 : {msync.ServiceName}') + self.out(f' 主服务器 : {msync.MasterURL}') + + # ─── 实验步骤 ────────────────────────────────────────── + + def _register_handlers(self): + from apps.sub_server.local_handlers import RegisterExperimentHandlers + from gvsdsdk.msync.local_tx import local_tx + + RegisterExperimentHandlers() + return all(local_tx.HasHandler(n) for n in ('demo_a', 'demo_b')) + + def _create_and_execute(self, msync, name, steps, timeout_seconds=120): + created = msync.TxCreate(name, steps, timeout_seconds=timeout_seconds) + if not created: + return None, 'TxCreate 失败' + tx_code = created.get('TransactionCode') + detail = msync.TxExecute(tx_code, access_token=msync.AccessToken) + if detail is None: + return tx_code, 'TxExecute 失败' + return tx_code, detail + + def _experiment_forward(self, msync): + tx_code, detail = self._create_and_execute(msync, '实验-正向', [ + {'name': 'demo_a', 'ParticipantService': msync.ServiceName, 'ActionPayload': {'in': 'a'}}, + {'name': 'demo_b', 'ParticipantService': msync.ServiceName, 'ActionPayload': {'in': 'b'}}, + ]) + if not detail: + return False + + self.out(f'\n 正向事务 {tx_code}: Status={detail.get("Status")} / Code={detail.get("StatusCode")}') + committed = detail.get('StatusCode') == 2 + if not committed: + self.out(f' 事务未提交: {detail}') + return False + + steps = {s['Index']: s for s in detail.get('Steps', [])} + step_a, step_b = steps.get(0), steps.get(1) + if not step_a or not step_b: + return False + + result_b = (step_b.get('Result') or {}).get('received_payload') or {} + flow_ok = bool(result_b.get('demo_a_uuid')) # demo_b 收到 demo_a 的输出 → 数据流生效 + self.out(f' 数据流验证: demo_b 收到 demo_a_uuid = {result_b.get("demo_a_uuid")}') + return committed and flow_ok + + def _experiment_compensate(self, msync): + tx_code, detail = self._create_and_execute(msync, '实验-补偿', [ + {'name': 'demo_a', 'ParticipantService': msync.ServiceName}, + {'name': 'demo_fail', 'ParticipantService': msync.ServiceName, 'MaxRetries': 0}, + ]) + if not detail: + return False + self.out(f'\n 补偿事务 {tx_code}: Status={detail.get("Status")} / Code={detail.get("StatusCode")}') + steps = {s['Index']: s for s in detail.get('Steps', [])} + return detail.get('StatusCode') == 4 and steps.get(0, {}).get('Status') in ('已补偿', 'Compensated') + + def _experiment_retry(self, msync): + self._reset_flaky_state() + tx_code, detail = self._create_and_execute(msync, '实验-重试', [ + {'name': 'demo_flaky', 'ParticipantService': msync.ServiceName, 'MaxRetries': 2}, + ]) + if not detail: + return False + self.out(f'\n 重试事务 {tx_code}: Status={detail.get("Status")} / Code={detail.get("StatusCode")}') + steps = {s['Index']: s for s in detail.get('Steps', [])} + step = steps.get(0, {}) + ok = detail.get('StatusCode') == 2 and step.get('RetryCount', 0) >= 1 + self.out(f' 自动重试次数: {step.get("RetryCount")}（要求 >=1）') + return ok + + def _reset_flaky_state(self): + """重置 runserver 进程内的 demo_flaky 计数器（通过 local/reset 端点）""" + import requests + try: + reset_url = f'{settings.MSYNC_SERVICE_DOMAIN}/api/msync/local/reset/' + requests.post(reset_url, timeout=5) + except Exception: + pass + + def _experiment_idempotency(self): + from gvsdsdk.msync.local_tx import local_tx + + r1 = local_tx.ExecuteStep('no_such_demo', {}, 'EXP:1:0') + r2 = local_tx.ExecuteStep('no_such_demo', {}, 'EXP:1:0') + r3 = local_tx.ExecuteStep('no_such_demo', {}, 'EXP:1:1') + return r1 is r2 and r1 is not r3 + + # ─── 汇报 ────────────────────────────────────────────── + + def _summary(self): + self.out('\n' + '=' * 60) + self.out('实验汇报') + self.out('=' * 60) + passed = sum(1 for _, ok, _ in self.results if ok) + for name, ok, detail in self.results: + mark = '✓' if ok else '✗' + self.out(f' {mark} {name}' + (f' ({detail})' if detail else '')) + self.out(f'\n结果: {passed}/{len(self.results)} 通过') + if passed != len(self.results): + self.stdout.write(self.style.ERROR('存在未通过的实验环节，请查看上方明细')) + raise SystemExit(1) diff --git a/ATestCServer/apps/sub_server/views.py b/ATestCServer/apps/sub_server/views.py index ab4daf48..83457fc7 100644 --- a/ATestCServer/apps/sub_server/views.py +++ b/ATestCServer/apps/sub_server/views.py @@ -397,4 +397,16 @@ class LocalHandlersView(APIView): def get(self, request): from gvsdsdk.msync.local_tx import local_tx - return Response(_ok(local_tx.ListHandlers())) \ No newline at end of file + return Response(_ok(local_tx.ListHandlers())) + + +class LocalResetView(APIView): + """重置实验处理器状态（demo_flaky 重试计数清零），仅限本地回环使用""" + + authentication_classes = [] + permission_classes = [AllowAny] + + def post(self, request): + from .local_handlers import ResetExperimentState + ResetExperimentState() + return Response(_ok({'reset': True})) \ No newline at end of file diff --git a/Main.py b/Main.py index 15df77b8..f19336c0 100644 --- a/Main.py +++ b/Main.py @@ -373,7 +373,9 @@ class SecurityMiddleware: # 非法HOST + 黑名单IP x_forwarded_for = self.get_ip(environ) - if (host not in WHITE_HOSTS and domain_suffix not in WHITE_HOSTS) or x_forwarded_for in BAN_IPS: + host_no_port = host.split(':')[0] if ':' in host else host + if (host not in WHITE_HOSTS and host_no_port not in WHITE_HOSTS + and domain_suffix not in WHITE_HOSTS) or x_forwarded_for in BAN_IPS: try: content = open(Path("TempLate") / "Errors" / "_403_HOST.html", encoding="utf-8").read() except Exception: diff --git a/gvsdsdk/msync/migrations/0001_initial.py b/gvsdsdk/msync/migrations/0001_initial.py index cb397daf..10021727 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(on_delete=django.db.models.deletion.CASCADE, related_name='tokens', to='msync.servicenode')), + ('ServiceUUID', models.ForeignKey(db_column='ServiceUUID', 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(on_delete=django.db.models.deletion.CASCADE, related_name='owned_objects', to='msync.servicenode')), + ('OwnerServiceUUID', models.ForeignKey(db_column='OwnerServiceUUID', 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, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='initiated_dtx', to='msync.servicenode')), + ('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')), ], 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, 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')), + ('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')), ], options={ 'verbose_name': '事务步骤', diff --git a/gvsdsdk/msync/models.py b/gvsdsdk/msync/models.py index 9388d005..fc31d8c4 100644 --- a/gvsdsdk/msync/models.py +++ b/gvsdsdk/msync/models.py @@ -88,6 +88,7 @@ class ServiceToken(QModel): TokenUUID = UUIDField(primary_key=True, verbose_name='令牌UUID') ServiceUUID = models.ForeignKey( ServiceNode, on_delete=models.CASCADE, + db_column='ServiceUUID', related_name='tokens', verbose_name='服务UUID', ) AccessToken = models.CharField(max_length=512, unique=True, verbose_name='访问令牌') @@ -128,6 +129,7 @@ class ObjectGlobalMapping(QModel): ObjectGlobalID = models.CharField(max_length=128, unique=True, db_index=True, verbose_name='全局ID') OwnerServiceUUID = models.ForeignKey( ServiceNode, on_delete=models.CASCADE, + db_column='OwnerServiceUUID', related_name='owned_objects', verbose_name='所属服务UUID', ) ObjectMetadata = models.JSONField(default=dict, verbose_name='对象元数据') @@ -198,7 +200,8 @@ class DistributedTransaction(QModel): ) InitiatorServiceUUID = models.ForeignKey( ServiceNode, on_delete=models.SET_NULL, - null=True, blank=True, related_name='initiated_dtx', verbose_name='发起服务UUID', + null=True, blank=True, db_column='InitiatorServiceUUID', + related_name='initiated_dtx', verbose_name='发起服务UUID', ) ReferenceOGMUUID = UUIDField(null=True, blank=True, verbose_name='关联OGM UUID') TransactionPayload = models.JSONField(default=dict, verbose_name='事务载荷') @@ -269,6 +272,7 @@ class TransactionStep(QModel): StepUUID = UUIDField(primary_key=True, verbose_name='步骤UUID') TransactionUUID = models.ForeignKey( DistributedTransaction, on_delete=models.CASCADE, + db_column='TransactionUUID', related_name='steps', verbose_name='事务UUID', ) StepIndex = models.IntegerField(verbose_name='步骤序号') @@ -276,7 +280,8 @@ class TransactionStep(QModel): StepDescription = models.CharField(max_length=512, default='', verbose_name='步骤描述') ParticipantServiceUUID = models.ForeignKey( ServiceNode, on_delete=models.SET_NULL, - null=True, blank=True, related_name='participated_steps', verbose_name='参与服务UUID', + null=True, blank=True, db_column='ParticipantServiceUUID', + related_name='participated_steps', verbose_name='参与服务UUID', ) StepStatus = models.SmallIntegerField( default=STEP_STATUS_PENDING, choices=STEP_STATUS_CHOICES, verbose_name='步骤状态',