perf: 删除 Key 前显式清理关联表，避免 CASCADE 级联删除超时

在 ProviderAPIKey 删除前，先批量删除 RequestCandidate、GeminiFileMapping、
VideoTask 等关联表记录，替代依赖数据库 CASCADE 级联删除，防止大量关联
记录导致删除操作超时。Pool 批量删除、封禁清理、Endpoint 批量删除三处
统一使用 cleanup_key_references。

src/api/admin/pool/routes.py

@@ -40,6 +40,7 @@ from src.services.provider.pool.scheduling_dimensions import (
     evaluate_pool_scheduling_dimensions,
     summarize_pool_scheduling_dimensions,
 )
+from src.services.provider_keys.key_side_effects import cleanup_key_references
 from src.services.provider_keys.quota_reader import get_quota_reader
 
 from .schemas import (
@@ -1071,12 +1072,18 @@ class AdminBatchActionKeysAdapter(AdminApiAdapter):
         if self.body.action == "delete":
             delete_started_at = time.perf_counter()
             sql_delete_ms = 0.0
+            cleanup_ms = 0.0
             commit_ms = 0.0
             side_effects_ms = 0.0
             key_ids = list(dict.fromkeys(self.body.key_ids))
             delete_batch_size = _resolve_delete_batch_size(db)
             delete_batch_count = 0
             try:
+                # 先清理关联表，避免 CASCADE 级联删除导致超时
+                cleanup_started_at = time.perf_counter()
+                cleanup_key_references(db, key_ids)
+                cleanup_ms = (time.perf_counter() - cleanup_started_at) * 1000.0
+
                 for batch in _iter_batches(key_ids, delete_batch_size):
                     batch_started_at = time.perf_counter()
                     result = db.execute(
@@ -1097,12 +1104,13 @@ class AdminBatchActionKeysAdapter(AdminApiAdapter):
                 db.rollback()
                 total_ms = (time.perf_counter() - delete_started_at) * 1000.0
                 logger.error(
-                    "batch delete commit failed: {} | provider={} requested={} batches={} sql_ms={:.2f} commit_ms={:.2f} total_ms={:.2f}",
+                    "batch delete commit failed: {} | provider={} requested={} batches={} sql_ms={:.2f} cleanup_ms={:.2f} commit_ms={:.2f} total_ms={:.2f}",
                     exc,
                     pid[:8],
                     len(key_ids),
                     delete_batch_count,
                     sql_delete_ms,
+                    cleanup_ms,
                     commit_ms,
                     total_ms,
                 )
@@ -1127,12 +1135,13 @@ class AdminBatchActionKeysAdapter(AdminApiAdapter):
 
             total_ms = (time.perf_counter() - delete_started_at) * 1000.0
             logger.info(
-                "[POOL_BATCH_DELETE_TIMING] provider={} requested={} affected={} batches={} batch_size={} sql_ms={:.2f} commit_ms={:.2f} side_effects_ms={:.2f} total_ms={:.2f}",
+                "[POOL_BATCH_DELETE_TIMING] provider={} requested={} affected={} batches={} batch_size={} cleanup_ms={:.2f} sql_ms={:.2f} commit_ms={:.2f} side_effects_ms={:.2f} total_ms={:.2f}",
                 pid[:8],
                 len(key_ids),
                 affected,
                 delete_batch_count,
                 delete_batch_size,
+                cleanup_ms,
                 sql_delete_ms,
                 commit_ms,
                 side_effects_ms,
@@ -1245,6 +1254,7 @@ class AdminCleanupBannedKeysAdapter(AdminApiAdapter):
 
         banned_key_ids = [str(key.id) for key in banned_keys]
         try:
+            cleanup_key_references(db, banned_key_ids)
             db.execute(
                 sa_delete(ProviderAPIKey).where(
                     ProviderAPIKey.provider_id == pid,

src/services/provider_keys/key_command_service.py

@@ -18,7 +18,10 @@ from src.core.crypto import crypto_service
 from src.core.exceptions import InvalidRequestException, NotFoundException
 from src.core.logger import logger
 from src.core.provider_types import ProviderType
-from src.models.database import Provider, ProviderAPIKey
+from src.models.database import (
+    Provider,
+    ProviderAPIKey,
+)
 from src.models.endpoint_models import (
     EndpointAPIKeyCreate,
     EndpointAPIKeyResponse,
@@ -28,6 +31,7 @@ from src.services.provider.fingerprint import generate_fingerprint, normalize_fi
 from src.services.provider_keys.auth_type import normalize_auth_type
 from src.services.provider_keys.duplicate_check import check_duplicate_key
 from src.services.provider_keys.key_side_effects import (
+    cleanup_key_references,
     run_create_key_side_effects,
     run_delete_key_side_effects,
     run_update_key_side_effects,
@@ -518,7 +522,10 @@ async def batch_delete_endpoint_keys_response(db: Session, key_ids: list[str]) -
     # 批量 SQL DELETE，一次提交
     success_count = 0
     try:
-        db.execute(sa_delete(ProviderAPIKey).where(ProviderAPIKey.id.in_(list(found_ids))))
+        found_id_list = list(found_ids)
+        # 先清理关联表，避免 CASCADE 级联删除超时
+        cleanup_key_references(db, found_id_list)
+        db.execute(sa_delete(ProviderAPIKey).where(ProviderAPIKey.id.in_(found_id_list)))
         db.commit()
         success_count = len(found_ids)
     except Exception as exc:

src/services/provider_keys/key_side_effects.py

@@ -4,13 +4,45 @@ Provider Key 写操作后的副作用处理。
 
 from __future__ import annotations
 
+from sqlalchemy import delete as sa_delete
 from sqlalchemy.orm import Session
 
 from src.api.base.models_service import invalidate_models_list_cache
 from src.core.logger import logger
-from src.models.database import ProviderAPIKey
+from src.models.database import GeminiFileMapping, ProviderAPIKey, RequestCandidate, VideoTask
 from src.services.cache.provider_cache import ProviderCacheService
 
+_SQLITE_BATCH_SIZE = 900
+_DEFAULT_BATCH_SIZE = 2000
+
+
+def cleanup_key_references(db: Session, key_ids: list[str]) -> None:
+    """在删除 ProviderAPIKey 前，先清理关联表记录，避免 CASCADE 级联删除超时。"""
+    if not key_ids:
+        return
+    batch_size = _resolve_batch_size(db)
+    for batch in _iter_batches(key_ids, batch_size):
+        db.execute(sa_delete(RequestCandidate).where(RequestCandidate.key_id.in_(batch)))
+        db.execute(sa_delete(GeminiFileMapping).where(GeminiFileMapping.key_id.in_(batch)))
+        db.execute(sa_delete(VideoTask).where(VideoTask.key_id.in_(batch)))
+
+
+def _resolve_batch_size(db: Session) -> int:
+    try:
+        bind = db.get_bind()
+        dialect_name = str(getattr(getattr(bind, "dialect", None), "name", "") or "").lower()
+    except Exception:
+        dialect_name = ""
+    if dialect_name == "sqlite":
+        return _SQLITE_BATCH_SIZE
+    return _DEFAULT_BATCH_SIZE
+
+
+def _iter_batches(items: list[str], batch_size: int) -> list[list[str]]:
+    if batch_size <= 0:
+        return [items]
+    return [items[i : i + batch_size] for i in range(0, len(items), batch_size)]
+
 
 async def run_update_key_side_effects(
     db: Session,