fix(provider_keys): 提升 Codex 配额异步同步的可靠性与可观测性

- 为 flush 引入 FlushResult，统一返回更新数与重试批次
- 增加指数退避与失败日志限流，成功后重置 backoff
- 批量提交失败时回退到单条提交，降低整批失败风险
- 补充测试，覆盖 flush 重试与提交失败回退场景

src/services/provider_keys/codex_quota_sync_dispatcher.py

@@ -10,6 +10,8 @@ Codex 配额实时同步调度器（异步去重版）。
 from __future__ import annotations
 
 import asyncio
+import time
+from dataclasses import dataclass
 from threading import Lock
 from typing import Any
 
@@ -20,16 +22,33 @@ from src.database.database import create_session
 from src.services.provider_keys.codex_realtime_quota import sync_codex_quota_from_response_headers
 
 
+@dataclass(slots=True)
+class FlushResult:
+    queued_count: int
+    updated_count: int
+    retry_batch: dict[str, dict[str, Any]]
+
+
 class CodexQuotaSyncDispatcher:
     """Codex 配额同步异步调度器。"""
 
-    def __init__(self, flush_interval_seconds: float = 0.5) -> None:
-        self.flush_interval_seconds = flush_interval_seconds
+    def __init__(
+        self,
+        flush_interval_seconds: float = 0.5,
+        *,
+        max_backoff_seconds: float = 8.0,
+        error_log_interval_seconds: float = 30.0,
+    ) -> None:
+        self.flush_interval_seconds = max(float(flush_interval_seconds), 0.001)
+        self.max_backoff_seconds = max(float(max_backoff_seconds), self.flush_interval_seconds)
+        self.error_log_interval_seconds = max(float(error_log_interval_seconds), 0.0)
         self._pending: dict[str, dict[str, Any]] = {}
         self._pending_lock = Lock()
         self._event: asyncio.Event | None = None
         self._loop: asyncio.AbstractEventLoop | None = None
         self._task: asyncio.Task[None] | None = None
+        self._current_flush_delay_seconds = self.flush_interval_seconds
+        self._last_flush_error_log_at: float | None = None
         self._running = False
 
     async def start(self) -> None:
@@ -38,10 +57,13 @@ class CodexQuotaSyncDispatcher:
         self._loop = asyncio.get_running_loop()
         self._event = asyncio.Event()
         self._task = asyncio.create_task(self._run(), name="codex-quota-sync-dispatcher")
+        self._current_flush_delay_seconds = self.flush_interval_seconds
+        self._last_flush_error_log_at = None
         self._running = True
         logger.info(
-            "Codex 配额异步同步器已启动，flush_interval={}s",
+            "Codex 配额异步同步器已启动，flush_interval={}s, max_backoff={}s",
             self.flush_interval_seconds,
+            self.max_backoff_seconds,
         )
 
     async def stop(self) -> None:
@@ -110,14 +132,35 @@ class CodexQuotaSyncDispatcher:
         try:
             while True:
                 await event.wait()
-                await asyncio.sleep(self.flush_interval_seconds)
+                await asyncio.sleep(self._current_flush_delay_seconds)
                 batch = self._drain_pending()
                 if batch:
                     try:
-                        await asyncio.to_thread(self._flush_batch_sync, batch)
+                        result = await asyncio.to_thread(self._flush_batch_sync, batch)
                     except Exception as exc:
-                        logger.warning("Codex 配额异步同步器 flush 失败，将在下一轮重试: {}", exc)
                         self._merge_back_pending(batch)
+                        self._increase_backoff()
+                        self._log_flush_failure_with_rate_limit(
+                            queued_count=len(batch),
+                            retry_count=len(batch),
+                            error=exc,
+                        )
+                    else:
+                        if result.updated_count > 0:
+                            logger.debug(
+                                "异步同步 Codex 配额完成: queued_keys={}, updated_keys={}",
+                                result.queued_count,
+                                result.updated_count,
+                            )
+                        if result.retry_batch:
+                            self._merge_back_pending(result.retry_batch)
+                            self._increase_backoff()
+                            self._log_flush_failure_with_rate_limit(
+                                queued_count=result.queued_count,
+                                retry_count=len(result.retry_batch),
+                            )
+                        else:
+                            self._reset_backoff()
                 with self._pending_lock:
                     if not self._pending:
                         event.clear()
@@ -125,7 +168,12 @@ class CodexQuotaSyncDispatcher:
             batch = self._drain_pending()
             if batch:
                 try:
-                    await asyncio.to_thread(self._flush_batch_sync, batch)
+                    result = await asyncio.to_thread(self._flush_batch_sync, batch)
+                    if result.retry_batch:
+                        logger.warning(
+                            "Codex 配额异步同步器停止时仍有未同步事件: retry_keys={}",
+                            len(result.retry_batch),
+                        )
                 except Exception as exc:
                     logger.warning("Codex 配额异步同步器停止时 flush 失败: {}", exc)
             raise
@@ -144,38 +192,114 @@ class CodexQuotaSyncDispatcher:
         with self._pending_lock:
             self._pending.update(batch)
 
-    def _flush_batch_sync(self, batch: dict[str, dict[str, Any]]) -> None:
-        if not batch:
+    def _reset_backoff(self) -> None:
+        self._current_flush_delay_seconds = self.flush_interval_seconds
+
+    def _increase_backoff(self) -> None:
+        self._current_flush_delay_seconds = min(
+            self.max_backoff_seconds,
+            max(self.flush_interval_seconds, self._current_flush_delay_seconds * 2),
+        )
+
+    def _log_flush_failure_with_rate_limit(
+        self,
+        *,
+        queued_count: int,
+        retry_count: int,
+        error: Exception | None = None,
+    ) -> None:
+        now = time.monotonic()
+        if (
+            self._last_flush_error_log_at is not None
+            and now - self._last_flush_error_log_at < self.error_log_interval_seconds
+        ):
             return
+        self._last_flush_error_log_at = now
+        if error is not None:
+            logger.warning(
+                "Codex 配额异步同步器 flush 失败，将重试: queued_keys={}, retry_keys={}, backoff={}s, error={}",
+                queued_count,
+                retry_count,
+                round(self._current_flush_delay_seconds, 3),
+                error,
+            )
+            return
+        logger.warning(
+            "Codex 配额异步同步器 flush 部分失败，将重试: queued_keys={}, retry_keys={}, backoff={}s",
+            queued_count,
+            retry_count,
+            round(self._current_flush_delay_seconds, 3),
+        )
+
+    def _flush_batch_fallback(
+        self,
+        entries: list[tuple[str, dict[str, Any]]],
+    ) -> tuple[int, dict[str, dict[str, Any]]]:
+        updated_count = 0
+        retry_batch: dict[str, dict[str, Any]] = {}
+        for provider_api_key_id, response_headers in entries:
+            db: Session = create_session()
+            try:
+                updated = sync_codex_quota_from_response_headers(
+                    db=db,
+                    provider_api_key_id=provider_api_key_id,
+                    response_headers=response_headers,
+                )
+                if updated:
+                    db.commit()
+                    updated_count += 1
+                else:
+                    db.rollback()
+            except Exception:
+                db.rollback()
+                retry_batch[provider_api_key_id] = response_headers
+            finally:
+                db.close()
+        return updated_count, retry_batch
+
+    def _flush_batch_sync(self, batch: dict[str, dict[str, Any]]) -> FlushResult:
+        if not batch:
+            return FlushResult(
+                queued_count=0,
+                updated_count=0,
+                retry_batch={},
+            )
 
         db: Session = create_session()
-        updated_count = 0
+        updated_entries: list[tuple[str, dict[str, Any]]] = []
+        retry_batch: dict[str, dict[str, Any]] = {}
         try:
             for provider_api_key_id, response_headers in batch.items():
                 try:
-                    updated = sync_codex_quota_from_response_headers(
-                        db=db,
-                        provider_api_key_id=provider_api_key_id,
-                        response_headers=response_headers,
-                    )
+                    with db.begin_nested():
+                        updated = sync_codex_quota_from_response_headers(
+                            db=db,
+                            provider_api_key_id=provider_api_key_id,
+                            response_headers=response_headers,
+                        )
                     if updated:
-                        db.commit()
-                        updated_count += 1
-                    else:
-                        db.rollback()
-                except Exception as exc:
+                        updated_entries.append((provider_api_key_id, response_headers))
+                except Exception:
+                    retry_batch[provider_api_key_id] = response_headers
+
+            updated_count = 0
+            if updated_entries:
+                try:
+                    db.commit()
+                    updated_count = len(updated_entries)
+                except Exception:
                     db.rollback()
-                    logger.warning(
-                        "异步同步 Codex 配额失败，已跳过: provider_api_key_id={}, error={}",
-                        provider_api_key_id,
-                        exc,
+                    fallback_updated, fallback_retry_batch = self._flush_batch_fallback(
+                        updated_entries
                     )
-            if updated_count > 0:
-                logger.debug(
-                    "异步同步 Codex 配额完成: queued_keys={}, updated_keys={}",
-                    len(batch),
-                    updated_count,
-                )
+                    updated_count = fallback_updated
+                    retry_batch.update(fallback_retry_batch)
+
+            return FlushResult(
+                queued_count=len(batch),
+                updated_count=updated_count,
+                retry_batch=retry_batch,
+            )
         finally:
             db.close()