Merge branch 'dev'

core/apk_manager.py

@@ -22,9 +22,11 @@
 import os
 import time
 import uuid
+
 import threading
 import subprocess
-from concurrent.futures import ThreadPoolExecutor, as_completed
+from concurrent.futures import (ThreadPoolExecutor, as_completed,
+                                TimeoutError as FuturesTimeout)
 
 from config import APK_DIR, ADB_PATH
 from core.logger import get_logger
@@ -35,6 +37,9 @@ _log = get_logger("core.apk")
 
 # adb install 超时（秒）。大 APK 安装慢，给 5 分钟
 _INSTALL_TIMEOUT = 300
+# 分块推送"多久没有任何字节被接收"就算链路卡死（秒）。注意这是**停滞**判据，
+# 不是总时长：正常推送会不断刷新它。
+_PUSH_STALL_TIMEOUT = 90
 # 并发安装数
 _INSTALL_CONCURRENCY = 5
 
@@ -333,7 +338,10 @@ class ApkManager:
                   f"跳过={len(serials)-len(install_list)}")
 
         if install_list:
-            with ThreadPoolExecutor(max_workers=_INSTALL_CONCURRENCY) as pool:
+            # 不用 with（它退出时会 wait 卡住的线程）：超时后要能立刻收尾，
+            # 否则界面永远停在"安装中"，后续安装全被"已有安装任务在进行"挡住
+            pool = ThreadPoolExecutor(max_workers=_INSTALL_CONCURRENCY)
+            try:
                 futures = {pool.submit(self._install_one, s, apk_path,
                                        package_name): s
                            for s in install_list}
@@ -343,8 +351,23 @@ class ApkManager:
                         fut.result()
                     except Exception as e:
                         self._set_item(s, "failed", f"安装异常: {e}")
+            except FuturesTimeout:
+                _log.warning(f"安装任务 {apk_name}: 有设备超时未返回，先收尾")
+                for s in install_list:
+                    it = ((self._install_task or {}).get("items") or {}).get(s) or {}
+                    if it.get("status") in ("pending", "installing"):
+                        self._set_item(s, "failed", "安装超时（设备无响应）")
+            except Exception as e:
+                _log.exception(f"安装任务 {apk_name} 异常")
+                for s in install_list:
+                    it = ((self._install_task or {}).get("items") or {}).get(s) or {}
+                    if it.get("status") in ("pending", "installing"):
+                        self._set_item(s, "failed", f"安装异常: {e}")
+            finally:
+                pool.shutdown(wait=False)
 
-        # 标记完成
+        # 标记完成（**必须**落到这里：没置 finished 的安装任务会一直挂着，
+        # 让"已有安装任务正在进行"永久挡住后续安装 —— 2026-09-14 事故）
         if self._install_task:
             self._install_task["finished"] = True
         # 统计
@@ -498,6 +521,33 @@ class ApkManager:
             _log.warning(f"[{serial}] exec-in 启动失败，退回 adb push: {e}")
             return self._push_fallback(serial, apk_path, remote)
 
+        # 看门狗：链路卡死时 `proc.stdin.write()` 会**永久阻塞**（设备掉线、
+        # WiFi 断了、任务在抢 adb…）→ 线程挂死 → 整批安装永不 finished → 界面
+        # 卡在"安装中"、后续安装全被"已有安装任务在进行"挡住（2026-09-14 事故）。
+        # 这里盯住"已写字节数"是否还在推进，停滞超时就 kill 掉 adb —— 阻塞中的
+        # write 会立刻以异常退出，走下面的回退逻辑。
+        # 为什么用看门狗而不是非阻塞写：`os.set_blocking`/管道 select 是 Unix 专有，
+        # 开发机是 Windows，看门狗两边都能用。
+        stop_watch = threading.Event()
+
+        def _stall_watchdog():
+            last_seen, since = -1, time.time()
+            while not stop_watch.wait(2.0):
+                if sent == total:
+                    return
+                if sent != last_seen:
+                    last_seen, since = sent, time.time()
+                elif time.time() - since > _PUSH_STALL_TIMEOUT:
+                    _log.warning(f"[{serial}] 推送停滞 {int(time.time() - since)} 秒"
+                                 f"（{sent}/{total} 字节），杀掉推送进程并回退")
+                    try:
+                        proc.kill()
+                    except Exception:
+                        pass
+                    return
+
+        watchdog = threading.Thread(target=_stall_watchdog, daemon=True)
+        watchdog.start()
         try:
             with open(apk_path, "rb") as f:
                 while True:
@@ -523,7 +573,11 @@ class ApkManager:
                 proc.kill()
             except Exception:
                 pass
-            return False, f"推送中断: {e}"
+            _log.warning(f"[{serial}] exec-in 推送中断（{e}），已传 {sent}/{total} 字节，"
+                         f"退回 adb push 重传")
+            return self._push_fallback(serial, apk_path, remote)
+        finally:
+            stop_watch.set()
 
         try:
             rc = proc.wait(timeout=_INSTALL_TIMEOUT)