From 4e51e348a53bfc5be6ed5431c1a8b64f60b4b7c4 Mon Sep 17 00:00:00 2001
From: cft0808 <cft0808@163.com>
Date: Sun, 5 Apr 2026 14:51:34 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E4=BB=BB=E5=8A=A1?=
 =?UTF-8?q?=E5=8D=A1=E6=AD=BB=E4=B8=89=E5=A4=A7=E9=97=AE=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

1. Gateway检测增加重试机制（3次+递增等待），避免瞬时不可达直接放弃
2. 新增定时巡检线程（每120秒），自动发现停滞任务并触发重试/升级/回滚
3. 回滚增加次数上限（最多3次），超限自动标记Blocked防止无限循环
---
 dashboard/server.py | 46 +++++++++++++++++++++++++++++++++++++++++----
 1 file changed, 42 insertions(+), 4 deletions(-)

diff --git a/dashboard/server.py b/dashboard/server.py
index 4a9c111..6b82912 100644
--- a/dashboard/server.py
+++ b/dashboard/server.py
@@ -1037,6 +1037,7 @@ def _scheduler_mark_progress(task, note=''):
     sched['stallSince'] = None
     sched['retryCount'] = 0
     sched['escalationLevel'] = 0
+    sched['rollbackCount'] = 0
     sched['lastEscalatedAt'] = None
     if note:
         _scheduler_add_flow(task, f'进展确认：{note}')
@@ -1219,9 +1220,21 @@ def handle_scheduler_scan(threshold_sec=600):
             continue
 
         if sched.get('autoRollback', True):
+            rollback_count = int(sched.get('rollbackCount') or 0)
+            max_rollback = int(sched.get('maxRollback') or 3)
             snapshot = sched.get('snapshot') or {}
             snap_state = snapshot.get('state')
-            if snap_state and snap_state != state:
+            if rollback_count >= max_rollback:
+                # 已达最大回滚次数，标记 Blocked 避免无限循环
+                if state != 'Blocked':
+                    task['state'] = 'Blocked'
+                    task['now'] = f'🚫 连续回滚{rollback_count}次仍无法推进，已自动挂起'
+                    task['block'] = f'连续停滞且回滚{rollback_count}次均失败，需人工介入'
+                    sched['stallSince'] = None
+                    _scheduler_add_flow(task, f'连续回滚{rollback_count}次，自动挂起等待人工介入')
+                    actions.append({'taskId': task_id, 'action': 'blocked', 'reason': f'max rollback {rollback_count}'})
+                    changed = True
+            elif snap_state and snap_state != state:
                 old_state = state
                 task['state'] = snap_state
                 task['org'] = snapshot.get('org', task.get('org', ''))
@@ -1229,9 +1242,10 @@ def handle_scheduler_scan(threshold_sec=600):
                 task['block'] = '无'
                 sched['retryCount'] = 0
                 sched['escalationLevel'] = 0
+                sched['rollbackCount'] = rollback_count + 1
                 sched['stallSince'] = None
                 sched['lastProgressAt'] = now_iso()
-                _scheduler_add_flow(task, f'连续停滞，自动回滚：{old_state} → {snap_state}')
+                _scheduler_add_flow(task, f'连续停滞，自动回滚：{old_state} → {snap_state}（第{rollback_count + 1}次）')
                 pending_rollbacks.append((task_id, snap_state))
                 actions.append({'taskId': task_id, 'action': 'rollback', 'toState': snap_state})
                 changed = True
@@ -2043,8 +2057,17 @@ def dispatch_for_state(task_id, task, new_state, trigger='state-transition'):
 
     def _do_dispatch():
         try:
-            if not _check_gateway_alive():
-                log.warning(f'⚠️ {task_id} 自动派发跳过: Gateway 未启动')
+            # Gateway 可能暂时不可达（休眠恢复、进程重启），等待后重试
+            import time as _time
+            _gw_alive = False
+            for _gw_attempt in range(3):
+                if _check_gateway_alive():
+                    _gw_alive = True
+                    break
+                if _gw_attempt < 2:
+                    _time.sleep(5 * (_gw_attempt + 1))  # 5s, 10s
+            if not _gw_alive:
+                log.warning(f'⚠️ {task_id} 自动派发跳过: Gateway 未启动（重试3次仍不可达）')
                 _update_task_scheduler(task_id, lambda t, s: s.update({
                     'lastDispatchAt': now_iso(),
                     'lastDispatchStatus': 'gateway-offline',
@@ -2752,6 +2775,21 @@ def main():
     # 启动恢复：重新派发上次被 kill 中断的 queued 任务
     threading.Timer(3.0, _startup_recover_queued_dispatches).start()
 
+    # 定时巡检：每 120 秒自动扫描停滞任务并触发重试/升级/回滚
+    def _periodic_scheduler_scan():
+        while True:
+            try:
+                import time as _time
+                _time.sleep(120)
+                result = handle_scheduler_scan(threshold_sec=180)
+                count = result.get('count', 0) if isinstance(result, dict) else 0
+                if count > 0:
+                    log.info(f'🔍 定时巡检：{count} 个动作')
+            except Exception as e:
+                log.warning(f'定时巡检异常: {e}')
+    threading.Thread(target=_periodic_scheduler_scan, daemon=True).start()
+    log.info('🔍 定时巡检已启动（每120秒）')
+
     try:
         server.serve_forever()
     except KeyboardInterrupt: