perf: 优化批量删除策略，缩小事务粒度并增强容错

- 批量删除分批大小从 500 降至 50，减少单事务锁持有时间
- 单批失败时跳过并继续，不再中断整个删除任务
- 进度上报改为按批次计数（每 5 批或末批），替代时间限频
- 关联表清理简化为直接 DELETE WHERE IN，移除逐行分批删除

src/services/provider_keys/batch_delete_task.py

@@ -8,7 +8,6 @@ from __future__ import annotations
 
 import asyncio
 import json
-import time
 import uuid
 from collections.abc import Callable
 
@@ -27,8 +26,8 @@ STATUS_FAILED = "failed"
 # 任务完成后保留时长（秒），也用作 Redis key 的 TTL
 _TASK_RETAIN_SECONDS = 600
 
-# 删除 ProviderAPIKey 时的分批大小
+# 批量删除时每个独立事务处理的 key 数量
-_DELETE_BATCH_SIZE = 500
+_CLEANUP_BATCH_SIZE = 50
 
 # Redis key 前缀
 _REDIS_KEY_PREFIX = "batch_delete_task"
@@ -175,7 +174,8 @@ def _sync_delete(
 ) -> int:
     """在线程中执行的同步删除逻辑，避免阻塞事件循环。
 
-    每批 cleanup + DELETE + commit 作为独立事务，减少锁持有时间。
+    按小批次（_CLEANUP_BATCH_SIZE）清理关联表并删除 key，
+    每个批次独立事务，单批失败跳过并继续，确保进度条持续推进。
     """
     from src.database import create_session
     from src.models.database import ProviderAPIKey
@@ -184,8 +184,11 @@ def _sync_delete(
     db = create_session()
     try:
         affected = 0
-        for i in range(0, len(key_ids), _DELETE_BATCH_SIZE):
+        total_batches = (len(key_ids) + _CLEANUP_BATCH_SIZE - 1) // _CLEANUP_BATCH_SIZE
-            batch = key_ids[i : i + _DELETE_BATCH_SIZE]
+        batch_idx = 0
+        for i in range(0, len(key_ids), _CLEANUP_BATCH_SIZE):
+            batch = key_ids[i : i + _CLEANUP_BATCH_SIZE]
+            batch_idx += 1
             try:
                 cleanup_key_references(db, batch)
                 result = db.execute(
@@ -197,14 +200,19 @@ def _sync_delete(
                 rowcount = getattr(result, "rowcount", 0) or 0
                 affected += int(rowcount)
                 db.commit()
-            except Exception:
+            except Exception as exc:
+                logger.warning(
+                    "[BATCH_DELETE] batch failed (keys {}-{}): {}",
+                    i,
+                    i + len(batch),
+                    exc,
+                )
                 try:
                     db.rollback()
                 except Exception:
                     pass
-                raise
+            # 每 5 个批次或最后一个批次上报进度，避免过于频繁写 Redis
-            # 通过回调通知进度（在事件循环线程中更新 Redis）
+            if progress_callback is not None and (batch_idx % 5 == 0 or batch_idx == total_batches):
-            if progress_callback is not None:
                 progress_callback(affected)
         return affected
     finally:
@@ -222,16 +230,10 @@ async def _run_batch_delete(
     r = await get_redis_client(require_redis=False)
     await _update_task_field(task_id, r=r, status=STATUS_RUNNING)
 
-    # 用于从工作线程安全地触发 Redis 进度更新（按时间限频，最多每 2 秒写一次）
+    # 从工作线程安全地触发 Redis 进度更新
     loop = asyncio.get_running_loop()
-    last_report_time = [0.0]
-    _PROGRESS_INTERVAL = 2.0
 
     def on_progress(current: int) -> None:
-        now = time.monotonic()
-        if now - last_report_time[0] < _PROGRESS_INTERVAL:
-            return
-        last_report_time[0] = now
         try:
             asyncio.run_coroutine_threadsafe(
                 _update_task_field(task_id, r=r, deleted=current), loop

src/services/provider_keys/key_side_effects.py

@@ -4,10 +4,7 @@ Provider Key 写操作后的副作用处理。
 
 from __future__ import annotations
 
-from typing import Any
-
 from sqlalchemy import delete as sa_delete
-from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from src.api.base.models_service import invalidate_models_list_cache
@@ -18,43 +15,22 @@ from src.services.cache.provider_cache import ProviderCacheService
 _SQLITE_BATCH_SIZE = 900
 _DEFAULT_BATCH_SIZE = 2000
 
-# RequestCandidate 等大表每轮 DELETE 的行数上限，
-# 避免单条 DELETE 语句扫描过多行导致超时。
-_ROW_DELETE_BATCH = 5000
-
 
 def cleanup_key_references(db: Session, key_ids: list[str]) -> None:
-    """在删除 ProviderAPIKey 前，先清理关联表记录，避免 CASCADE 级联删除超时。"""
+    """在删除 ProviderAPIKey 前，先清理关联表记录，避免 CASCADE 级联删除超时。
+
+    PostgreSQL 下直接使用 DELETE WHERE key_id IN (...)，利用 key_id 索引高效删除。
+    SQLite 下按 key_id 批次拆分，避免超出变量数限制。
+    """
     if not key_ids:
         return
     batch_size = _resolve_batch_size(db)
     for batch in _iter_batches(key_ids, batch_size):
-        # RequestCandidate 数据量通常最大，分批按行数删除
+        db.execute(sa_delete(RequestCandidate).where(RequestCandidate.key_id.in_(batch)))
-        _delete_in_row_batches(db, RequestCandidate, RequestCandidate.key_id, batch)
         db.execute(sa_delete(GeminiFileMapping).where(GeminiFileMapping.key_id.in_(batch)))
         db.execute(sa_delete(VideoTask).where(VideoTask.key_id.in_(batch)))
 
 
-def _delete_in_row_batches(db: Session, model: Any, fk_col: Any, key_id_batch: list[str]) -> int:
-    """分批删除大表记录，每次最多删除 _ROW_DELETE_BATCH 行，避免单条语句超时。"""
-    total = 0
-    while True:
-        row_ids = [
-            r[0]
-            for r in db.execute(
-                select(model.id).where(fk_col.in_(key_id_batch)).limit(_ROW_DELETE_BATCH)
-            )
-        ]
-        if not row_ids:
-            break
-        result = db.execute(sa_delete(model).where(model.id.in_(row_ids)))
-        deleted = getattr(result, "rowcount", 0) or 0
-        total += deleted
-        if len(row_ids) < _ROW_DELETE_BATCH:
-            break
-    return total
-
-
 def _resolve_batch_size(db: Session) -> int:
     try:
         bind = db.get_bind()