dd

domain-api/app/services/sync_push_service.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import socket
 import urllib.error
 import urllib.parse
 import urllib.request
@@ -8,6 +9,7 @@ from datetime import datetime, timedelta
 
 from app.core.config import settings
 from app.core.db import get_db
+from app.services.cluster_runtime_service import register_node_heartbeat
 from app.services.sync_record_service import _decode_json, _normalize_region
 
 
@@ -34,6 +36,38 @@ def _projection_ingest_type(sync_type: str) -> str:
     return "sync_ingest"
 
 
+def _refresh_remote_runtime_node(*, source_region: str, projection: dict, received_at: datetime | None = None) -> None:
+    node_info = projection.get("node") or {}
+    node_code = str(node_info.get("node_code") or "").strip()
+    region = _normalize_region(node_info.get("region"), source_region)
+    role = str(node_info.get("role") or "control").strip() or "control"
+    hostname = str(node_info.get("hostname") or "").strip() or socket.gethostname()
+    ip = str(node_info.get("ip") or "").strip()
+    if not node_code:
+        node_code = f"{region}-{role}-imported"
+
+    metadata = {
+        "service": "runtime-ingest",
+        "projection_source_region": source_region,
+        "worker_mode": projection.get("worker_mode", ""),
+        "phase_label": projection.get("phase_label", ""),
+        "phase_detail": projection.get("phase_detail", ""),
+        "proxy_runtime_label": projection.get("proxy_runtime_label", ""),
+        "proxy_runtime_reason": projection.get("proxy_runtime_reason", ""),
+        "updated_at": _format_time(received_at or datetime.now()),
+    }
+    register_node_heartbeat(
+        node_code=node_code,
+        region=region,
+        role=role,
+        status="online",
+        current_load=int(((projection.get("progress") or {}).get("running", 0) or 0)),
+        metadata=metadata,
+        hostname_override=hostname,
+        ip_override=ip,
+    )
+
+
 def _load_latest_projection(sync_type: str) -> dict | None:
     source_region = _normalize_region(settings.sync_source_region, settings.node_region)
     target_region = _normalize_region(settings.sync_target_region, "overseas")
@@ -213,6 +247,7 @@ def ingest_runtime_projection(payload: dict, *, shared_token: str | None = None)
     projection_hash = str(payload.get("projection_hash") or "").strip()
     projection = payload.get("projection") or {}
     target_region = _normalize_region(settings.node_region, "overseas")
+    received_at = datetime.now()
 
     with get_db() as conn:
         with conn.cursor() as cur:
@@ -232,6 +267,17 @@ def ingest_runtime_projection(payload: dict, *, shared_token: str | None = None)
             )
             existing = cur.fetchone()
             if existing:
+                if sync_type == "runtime_projection":
+                    _refresh_remote_runtime_node(source_region=source_region, projection=projection, received_at=received_at)
+                cur.execute(
+                    """
+                    UPDATE detect_sync_records
+                    SET updated_at = CURRENT_TIMESTAMP
+                    WHERE id = %s
+                    """,
+                    (int(existing[0]),),
+                )
+                conn.commit()
                 return True, "同步投影已存在，已按幂等处理", {"record_id": int(existing[0]), "deduplicated": True}
 
             stored_payload = {
@@ -239,7 +285,7 @@ def ingest_runtime_projection(payload: dict, *, shared_token: str | None = None)
                 "source_record_id": source_record_id,
                 "projection_hash": projection_hash,
                 "projection": projection,
-                "received_at": _format_time(datetime.now()),
+                "received_at": _format_time(received_at),
             }
             cur.execute(
                 """
@@ -259,6 +305,8 @@ def ingest_runtime_projection(payload: dict, *, shared_token: str | None = None)
             )
             record_id = int(cur.fetchone()[0])
         conn.commit()
+    if sync_type == "runtime_projection":
+        _refresh_remote_runtime_node(source_region=source_region, projection=projection, received_at=received_at)
     return True, "同步投影接收成功", {"record_id": record_id, "deduplicated": False}
 
 
@@ -313,12 +361,22 @@ def _push_projection_batch(sync_type: str, ingest_url: str) -> tuple[bool, str,
 def _push_projection_record(source_record: dict, sync_type: str, ingest_url: str) -> tuple[bool, str, dict]:
     latest_attempt = _latest_push_attempt(source_record["id"], source_record["target_region"], sync_type)
     if latest_attempt and latest_attempt["status"] == "success":
-        return True, "该投影已推送，无需重复发送", {
-            "action": "push_sync",
-            "sync_type": sync_type,
-            "source_record_id": source_record["id"],
-            "deduplicated": True,
-        }
+        last_created_at = latest_attempt.get("created_at")
+        if sync_type != "runtime_projection" or not isinstance(last_created_at, datetime):
+            return True, "该投影已推送，无需重复发送", {
+                "action": "push_sync",
+                "sync_type": sync_type,
+                "source_record_id": source_record["id"],
+                "deduplicated": True,
+            }
+        now = datetime.now(last_created_at.tzinfo) if last_created_at.tzinfo else datetime.now()
+        if now - last_created_at < timedelta(seconds=max(20, int(settings.sync_poll_interval_seconds or 30))):
+            return True, "该投影已推送，无需重复发送", {
+                "action": "push_sync",
+                "sync_type": sync_type,
+                "source_record_id": source_record["id"],
+                "deduplicated": True,
+            }
     if latest_attempt and latest_attempt["status"] == "pending":
         return True, "该投影已有同步推送进行中，暂不重复发送", {
             "action": "push_sync",