fix(msync): 修复主从链路验证暴露的缺陷，ATestCServer 自动化实验 6/6 通过

- Main.py: Host 白名单剥离端口再匹配，放行 127.0.0.1:8000 自省请求
- ATestCServer: 新增 /api/msync/local/* SDK 契约路由、实验处理器
  模块 local_handlers（随服务进程注册）、local/reset 状态重置端点；
  msync_experiment 命令 check 改名 record 并简化补偿/重试逻辑
- gvsdsdk msync 模型与迁移 0001 的 FK db_column 同步（与独立仓一致）

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'),
 ]

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

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

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)

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()))
+        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}))

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:

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': '事务步骤',

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='步骤状态',