refactor: 优化 pending 清理和压缩任务的内存使用

- 将 pending 请求清理改为分批处理，使用轻量列查询代替全量 ORM 加载
- 限制 pending 清理和历史压缩的批次大小上限，防止单次查询占用过多内存
- 移除 processed_ids 集合，改用 synchronize_session=False 避免内存累积
- 更新 README 中部署方式描述

README.md

@@ -34,7 +34,7 @@ Aether 是一个自托管的 AI API 网关，为团队和个人提供多租户
 
 ## 部署
 
-### Docker Compose（推荐：本地构建镜像）
+### Docker Compose（推荐：预构建镜像）
 
 ```bash
 # 1. 克隆代码

src/services/system/maintenance_scheduler.py

@@ -531,8 +531,15 @@ class MaintenanceScheduler:
                 timeout_minutes = SystemConfigService.get_config(
                     db, "pending_request_timeout_minutes", 10
                 )
+                # pending 清理涉及 candidate 表关联查询，限制批次大小以控制内存
+                batch_size = min(
+                    max(SystemConfigService.get_config(db, "cleanup_batch_size", 1000), 1),
+                    200,
+                )
                 return UsageService.cleanup_stale_pending_requests(
-                    db, timeout_minutes=timeout_minutes
+                    db,
+                    timeout_minutes=timeout_minutes,
+                    batch_size=batch_size,
                 )
             except Exception as e:
                 logger.exception(f"清理 pending 请求失败: {e}")
@@ -929,7 +936,7 @@ class MaintenanceScheduler:
 
         total_compressed = 0
         no_progress_count = 0
-        processed_ids: set = set()
+        memory_safe_batch_size = max(1, min(batch_size, 100))
 
         while True:
             batch_db = create_session()
@@ -949,7 +956,7 @@ class MaintenanceScheduler:
                         | (Usage.provider_request_body.isnot(None))
                         | (Usage.client_response_body.isnot(None))
                     )
-                    .limit(batch_size)
+                    .limit(memory_safe_batch_size)
                     .all()
                 )
 
@@ -983,15 +990,6 @@ class MaintenanceScheduler:
                     batch_db.commit()
                     continue
 
-                current_ids = {r.id for r in valid_records}
-                repeated_ids = current_ids & processed_ids
-                if repeated_ids:
-                    logger.error(
-                        f"检测到重复处理的记录 ID: {list(repeated_ids)[:5]}...，"
-                        "说明数据库更新未生效，终止循环"
-                    )
-                    break
-
                 batch_success = 0
 
                 for r in valid_records:
@@ -1021,10 +1019,10 @@ class MaintenanceScheduler:
                                     else None
                                 ),
                             )
+                            .execution_options(synchronize_session=False)
                         )
                         if result.rowcount > 0:
                             batch_success += 1
-                            processed_ids.add(r.id)
                     except Exception as e:
                         logger.warning(f"压缩记录 {r.id} 失败: {e}")
                         continue

src/services/usage/active_requests.py

@@ -104,6 +104,7 @@ class UsageActiveRequestsMixin:
         cls,
         db: Session,
         timeout_minutes: int = 10,
+        batch_size: int = 200,
     ) -> int:
         """
         清理超时的 pending/streaming 请求
@@ -115,6 +116,7 @@ class UsageActiveRequestsMixin:
         Args:
             db: 数据库会话
             timeout_minutes: 超时时间（分钟），默认 10 分钟
+            batch_size: 每次处理的记录数，限制在 1-200 之间
 
         Returns:
             清理的记录数
@@ -122,68 +124,80 @@ class UsageActiveRequestsMixin:
         now = datetime.now(timezone.utc)
         cutoff_time = now - timedelta(minutes=timeout_minutes)
 
-        # 查找超时的请求
+        batch_size = max(1, batch_size)
-        stale_requests = (
-            db.query(Usage)
-            .filter(
-                Usage.status.in_(["pending", "streaming"]),
-                Usage.created_at < cutoff_time,
-            )
-            .all()
-        )
-
-        if not stale_requests:
-            return 0
-
-        # 收集所有超时请求的 request_id，查询哪些实际已成功完成
-        stale_request_ids = [u.request_id for u in stale_requests if u.request_id]
-        completed_request_ids = cls._find_completed_request_ids(db, stale_request_ids)
-
         failed_count = 0
         recovered_count = 0
 
-        for usage in stale_requests:
+        while True:
-            old_status = usage.status
+            stale_requests = (
-            if usage.request_id and usage.request_id in completed_request_ids:
+                db.query(Usage.id, Usage.request_id, Usage.status)
-                # Provider 已返回成功，恢复为 completed
+                .filter(
-                usage.status = "completed"
+                    Usage.status.in_(["pending", "streaming"]),
-                usage.status_code = 200
+                    Usage.created_at < cutoff_time,
-                usage.error_message = None
-                recovered_count += 1
-            else:
-                # 无成功 candidate，标记为 failed
-                usage.status = "failed"
-                usage.error_message = (
-                    f"请求超时: 状态 '{old_status}' 超过 {timeout_minutes} 分钟未完成"
                 )
-                usage.status_code = 504
+                .order_by(Usage.created_at.asc(), Usage.id.asc())
-                failed_count += 1
+                .limit(batch_size)
-
+                .all()
-        # 同步更新成功请求的 candidate 状态：streaming -> success
-        cls._sync_candidate_status_to_success(db, list(completed_request_ids), now)
-
-        # 同步更新失败请求的 candidate 状态：streaming/pending -> failed
-        failed_request_ids = [
-            u.request_id
-            for u in stale_requests
-            if u.request_id and u.request_id not in completed_request_ids
-        ]
-        if failed_request_ids:
-            db.query(RequestCandidate).filter(
-                RequestCandidate.request_id.in_(failed_request_ids),
-                RequestCandidate.status.in_(["streaming", "pending"]),
-            ).update(
-                {
-                    "status": "failed",
-                    "finished_at": now,
-                    "error_message": "请求超时（服务器可能已重启）",
-                },
-                synchronize_session=False,
             )
 
+            if not stale_requests:
+                break
+
+            stale_request_ids = [request_id for _, request_id, _ in stale_requests if request_id]
+            completed_request_ids = cls._find_completed_request_ids(db, stale_request_ids)
+
+            usage_updates = []
+            failed_request_ids: list[str] = []
+
+            for usage_id, request_id, old_status in stale_requests:
+                if request_id and request_id in completed_request_ids:
+                    usage_updates.append(
+                        {
+                            "id": usage_id,
+                            "status": "completed",
+                            "status_code": 200,
+                            "error_message": None,
+                        }
+                    )
+                    recovered_count += 1
+                else:
+                    usage_updates.append(
+                        {
+                            "id": usage_id,
+                            "status": "failed",
+                            "status_code": 504,
+                            "error_message": (
+                                f"请求超时: 状态 '{old_status}' 超过 {timeout_minutes} 分钟未完成"
+                            ),
+                        }
+                    )
+                    failed_count += 1
+                    if request_id:
+                        failed_request_ids.append(request_id)
+
+            if usage_updates:
+                db.bulk_update_mappings(Usage, usage_updates)
+
+            cls._sync_candidate_status_to_success(db, list(completed_request_ids), now)
+
+            if failed_request_ids:
+                db.query(RequestCandidate).filter(
+                    RequestCandidate.request_id.in_(failed_request_ids),
+                    RequestCandidate.status.in_(["streaming", "pending"]),
+                ).update(
+                    {
+                        "status": "failed",
+                        "finished_at": now,
+                        "error_message": "请求超时（服务器可能已重启）",
+                    },
+                    synchronize_session=False,
+                )
+
+            db.commit()
+            db.expunge_all()
+
         total = failed_count + recovered_count
         if total > 0:
-            db.commit()
             parts = []
             if failed_count:
                 parts.append(f"{failed_count} 条标记为 failed")