tupian duxie ga dao duli xianchengchi, bing jiaru shijian xunhuan kadun tanzhen

api/app.py

@@ -1,11 +1,12 @@
 from __future__ import annotations
 
+import asyncio
 from contextlib import asynccontextmanager
 from html import escape
 from mimetypes import guess_type
 from pathlib import Path
 import re
-from threading import Event, Thread
+from threading import Event, Thread, active_count
 import time
 
 from fastapi import FastAPI, HTTPException, Request
@@ -156,6 +157,26 @@ def _web_response(full_path: str, request: Request, asset: Path) -> Response:
 def create_app() -> FastAPI:
     app_version = config.app_version
 
+    async def _loop_lag_watchdog() -> None:
+        """把事件循环的真实卡顿写进日志。
+
+        ⚠️ 排查「整站无响应」时，容器日志一切正常（后台线程照跑）却测不出来到底卡在哪，
+        就是因为没有任何关于事件循环本身的观测。这个探针每 20 秒量一次 sleep 漂移：
+        漂移 = 循环线程被同步调用/GIL 抢占的实际时长。
+        """
+        from utils.log import logger as _logger
+
+        while True:
+            started = time.perf_counter()
+            await asyncio.sleep(20)
+            lag_ms = round((time.perf_counter() - started - 20) * 1000)
+            if lag_ms >= 800:
+                _logger.warning({
+                    "event": "event_loop_lag",
+                    "lag_ms": lag_ms,
+                    "threads": active_count(),
+                })
+
     @asynccontextmanager
     async def lifespan(_: FastAPI):
         # anyio's worker-thread limiter defaults to 40 and is shared by every
@@ -171,9 +192,11 @@ def create_app() -> FastAPI:
         share_evidence_service.start()
         _start_bundled_skill_sync()
         _start_image_cleanup()
+        watchdog = asyncio.create_task(_loop_lag_watchdog())
         try:
             yield
         finally:
+            watchdog.cancel()
             stop_event.set()
             thread.join(timeout=1)
             backup_service.stop()

api/system.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+import asyncio
+import functools
 import json
+from concurrent.futures import ThreadPoolExecutor
 from typing import Any
 from urllib.parse import quote, urlparse
 
@@ -28,6 +31,32 @@ from services.auth_service import auth_service
 from services.image_service import delete_images, download_images_zip, get_image_download_response, get_image_response, get_thumbnail_response, list_images
 from services.image_storage_service import ImageStorageError, image_storage_service
 from services.image_tags_service import delete_tag, get_all_tags, set_tags
+
+# ⚠️⚠️ 图片读写必须走**独立线程池**，不能和 StaticFiles / 其他 run_in_threadpool 共用
+# AnyIO 的默认线程池（本服务设成 50）。原因：图片存在**网络挂载**的 Storage Bucket 上，
+# 单张 2.8MB 在桶上读一次可能要几十秒；几十人同时刷一屏图（每人 6~9 张）会把 50 个共享
+# 线程全部占满，后面排队的 HTML、JS、/health 全部超时 —— 表现是「连接被关闭 / 一个字节
+# 都不回」（2026-10-19 全站无响应事故，后台线程日志一切正常，就是 HTTP 不响应）。
+# 独立池 = 图片再慢也只排队在自己这里，页面和接口照常可用。
+_IMAGE_IO_POOL = ThreadPoolExecutor(max_workers=8, thread_name_prefix="image-io")
+# 单张图片在桶上读不完时的上限：超时直接 503 + Retry-After，让浏览器/CF 稍后重试。
+# ⚠️ 不能无限排队：几百个请求排在 8 个 worker 后面，会变成十几分钟的积压，
+# 用户看到的就是「页面一直在转」，而 CF 也会一直保持连接。
+_IMAGE_IO_TIMEOUT_SECONDS = 25.0
+
+
+async def _run_image_io(func, *args):
+    """在图片专用线程池里跑阻塞的图片读写，隔离出共享线程池。"""
+    loop = asyncio.get_running_loop()
+    future = loop.run_in_executor(_IMAGE_IO_POOL, functools.partial(func, *args))
+    try:
+        return await asyncio.wait_for(future, timeout=_IMAGE_IO_TIMEOUT_SECONDS)
+    except asyncio.TimeoutError:
+        raise HTTPException(
+            status_code=503,
+            detail={"error": "image io timeout"},
+            headers={"Retry-After": "10"},
+        )
 from services.proxy_service import proxy_settings, test_clearance, test_proxy
 from services.share_evidence_service import share_evidence_service
 from services.pool_pricing_service import pool_pricing_service
@@ -527,7 +556,7 @@ def create_router(app_version: str) -> APIRouter:
         # 线程上跑；而图片是全站最热的请求，几十人同时刷图会把事件循环彻底堵死 ⇒ 连 /health、
         # 静态资源和页面 SSR 都一起排队超时（2026-10-16 线上实测：/health 连续 20~25s 无响应，
         # 而同一 commit 的空闲容器 1s 返回 200）。
-        return await run_in_threadpool(get_image_response, image_path)
+        return await _run_image_io(get_image_response, image_path)
 
     @router.get("/api/images/proxy")
     async def proxy_image(url: str, authorization: str | None = Header(default=None)):
@@ -535,7 +564,7 @@ def create_router(app_version: str) -> APIRouter:
         parsed = urlparse(str(url or "").strip())
         if parsed.scheme not in {"http", "https"} or not parsed.netloc:
             raise HTTPException(status_code=400, detail={"error": "invalid image url"})
-        data, _filename, content_type = await run_in_threadpool(_download_image_url, url)
+        data, _filename, content_type = await _run_image_io(_download_image_url, url)
         return Response(
             content=data,
             media_type=content_type,
@@ -546,17 +575,18 @@ def create_router(app_version: str) -> APIRouter:
     async def get_image_thumbnail(image_path: str):
         # ⚠️ 同上：ensure_thumbnail 会读原图 + 用 PIL 做 LANCZOS 重采样并 PNG 编码，
         # 是纯 CPU 密集的阻塞活，放事件循环上跑会把所有请求一起拖死。
-        return await run_in_threadpool(get_thumbnail_response, image_path)
+        return await _run_image_io(get_thumbnail_response, image_path)
 
     @router.post("/api/images/delete")
     async def delete_images_endpoint(body: ImageDeleteRequest, authorization: str | None = Header(default=None)):
         require_admin(authorization)
-        return delete_images(body.paths, start_date=body.start_date.strip(), end_date=body.end_date.strip(), all_matching=body.all_matching)
+        return await _run_image_io(delete_images, body.paths, start_date=body.start_date.strip(), end_date=body.end_date.strip(), all_matching=body.all_matching)
 
     @router.post("/api/images/download")
     async def download_images_endpoint(body: ImageDownloadRequest, authorization: str | None = Header(default=None)):
         require_admin(authorization)
-        buf = download_images_zip(body.paths)
+        # 打包下载要逐张从网络 Bucket 读原图，阻塞时间比单张更长 —— 更不能占共享线程池。
+        buf = await _run_image_io(download_images_zip, body.paths)
         return StreamingResponse(
             buf,
             media_type="application/zip",
@@ -566,8 +596,8 @@ def create_router(app_version: str) -> APIRouter:
     @router.get("/api/images/download/{image_path:path}")
     async def download_single_image_endpoint(image_path: str, authorization: str | None = Header(default=None)):
         require_admin(authorization)
-        # ⚠️ 同样会把整张图从网络 Bucket 读进内存，必须进线程池。
-        return await run_in_threadpool(get_image_download_response, image_path)
+        # ⚠️ 同样会把整张图从网络 Bucket 读进内存，必须进**图片专用**线程池。
+        return await _run_image_io(get_image_download_response, image_path)
 
     @router.get("/api/logs")
     async def get_logs(type: str = "", start_date: str = "", end_date: str = "", limit: int = 200, authorization: str | None = Header(default=None)):
@@ -756,9 +786,16 @@ def create_router(app_version: str) -> APIRouter:
         # ⚠️ 健康检查要保证「容器越忙也测得出来」：这两个都是同步读库/探活的阻塞调用，
         # 直接放事件循环上跑的话，图片请求一多，/health 自己就先被堵死，反而失去判活能力
         # （2026-10-16 线上现象：/health 20~25s 超时，让人误以为容器已死）。
-        stats = await run_in_threadpool(acct_svc.get_stats)
+        # 再加一层硬超时：线程池被占满时也必须秒回，否则连"容器是否活着"都测不出来。
+        try:
+            stats = await asyncio.wait_for(run_in_threadpool(acct_svc.get_stats), timeout=3.0)
+        except Exception:
+            stats = {"active": 0, "total": 0, "unavailable_stats": True}
         storage = config.get_storage_backend()
-        storage_health = await run_in_threadpool(storage.health_check)
+        try:
+            storage_health = await asyncio.wait_for(run_in_threadpool(storage.health_check), timeout=3.0)
+        except Exception:
+            storage_health = {"ok": False, "error": "timeout"}
         healthy = int(stats.get("active") or 0) > 0
         payload = {
             "status": "ok" if healthy else "degraded",