perf: 同步 DB 操作迁移到 asyncio.to_thread，避免阻塞事件循环

failover/stream_telemetry/recording/stream 中的同步 DB 操作（commit/execute/query）
会阻塞 asyncio 事件循环，导致 Hub PING 心跳无法发送、worker idle timeout 断连。
将这些操作包装到 asyncio.to_thread() 中执行。

同时提取 Alembic 迁移脚本中重复的幂等性辅助函数到 alembic/helpers.py，
用批量查询缓存替代逐条 information_schema 查询，backfill SQL 合并为 LEFT JOIN。

新增 failover 中客户端断连的快速终止路径，避免继续无意义的重试。

alembic/helpers.py

@@ -0,0 +1,171 @@
+"""Shared idempotent helpers for Alembic migrations.
+
+All metadata lookups are batched: one query loads an entire table's column info
+or all FK delete rules, then results are cached for the migration's lifetime.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from collections.abc import Callable
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# ---------------------------------------------------------------------------
+# Batch metadata cache
+# ---------------------------------------------------------------------------
+
+
+class _SchemaCache:
+    """Lazy, per-migration cache for information_schema lookups.
+
+    Call ``load_columns(tables)`` / ``load_fk_rules(tables)`` once at the top
+    of ``upgrade()`` or ``downgrade()`` to prime the cache.  Subsequent
+    ``column_exists`` / ``column_type`` / ``fk_ondelete`` calls are pure
+    dict lookups -- zero extra DB round-trips.
+    """
+
+    def __init__(self) -> None:
+        # {table_name: {column_name: data_type}}
+        self._columns: dict[str, dict[str, str]] = {}
+        # {(table_name, constraint_name): delete_rule}
+        self._fk_rules: dict[tuple[str, str], str] = {}
+        self._fk_loaded_tables: set[str] = set()
+
+    # -- loaders (one query per call) --------------------------------------
+
+    def load_columns(self, tables: list[str]) -> None:
+        """Fetch column names + data types for *tables* in one query."""
+        need = [t for t in tables if t not in self._columns]
+        if not need:
+            return
+        bind = op.get_bind()
+        rows = bind.execute(
+            sa.text(
+                "SELECT table_name, column_name, data_type "
+                "FROM information_schema.columns "
+                "WHERE table_name = ANY(:tables)"
+            ),
+            {"tables": need},
+        ).fetchall()
+        # Initialise even empty tables so we don't re-query
+        for t in need:
+            self._columns.setdefault(t, {})
+        for table, col, dtype in rows:
+            self._columns[table][col] = dtype
+
+    def load_fk_rules(self, tables: list[str]) -> None:
+        """Fetch FK delete rules for *tables* in one query."""
+        need = [t for t in tables if t not in self._fk_loaded_tables]
+        if not need:
+            return
+        bind = op.get_bind()
+        rows = bind.execute(
+            sa.text(
+                "SELECT tc.table_name, tc.constraint_name, rc.delete_rule "
+                "FROM information_schema.referential_constraints rc "
+                "JOIN information_schema.table_constraints tc "
+                "  ON rc.constraint_name = tc.constraint_name "
+                "WHERE tc.table_name = ANY(:tables)"
+            ),
+            {"tables": need},
+        ).fetchall()
+        for table, name, rule in rows:
+            self._fk_rules[(table, name)] = rule
+        self._fk_loaded_tables.update(need)
+
+    # -- lookups (pure dict, zero DB) --------------------------------------
+
+    def column_exists(self, table: str, column: str) -> bool:
+        return column in self._columns.get(table, {})
+
+    def column_type(self, table: str, column: str) -> str | None:
+        return self._columns.get(table, {}).get(column)
+
+    def is_numeric(self, table: str, column: str) -> bool:
+        return self.column_type(table, column) == "numeric"
+
+    def fk_ondelete(self, table: str, constraint: str) -> str | None:
+        return self._fk_rules.get((table, constraint))
+
+    def invalidate_columns(self, table: str) -> None:
+        """Force re-load on next load_columns() for *table* (after ADD/DROP COLUMN)."""
+        self._columns.pop(table, None)
+
+
+def new_cache() -> _SchemaCache:
+    """Create a fresh schema cache for a single migration run."""
+    return _SchemaCache()
+
+
+# ---------------------------------------------------------------------------
+# Idempotent DDL helpers
+# ---------------------------------------------------------------------------
+
+
+def replace_fk_if_needed(
+    cache: _SchemaCache,
+    constraint_name: str,
+    table_name: str,
+    ref_table: str,
+    local_cols: list[str],
+    remote_cols: list[str],
+    desired_ondelete: str,
+) -> None:
+    """Drop and recreate a FK only if the current ON DELETE rule differs."""
+    current = cache.fk_ondelete(table_name, constraint_name)
+    if current and current.upper() == desired_ondelete.upper():
+        return
+    if current:
+        op.drop_constraint(constraint_name, table_name, type_="foreignkey")
+    op.create_foreign_key(
+        constraint_name,
+        table_name,
+        ref_table,
+        local_cols,
+        remote_cols,
+        ondelete=desired_ondelete,
+    )
+
+
+def index_exists(index_name: str) -> bool:
+    bind = op.get_bind()
+    result = bind.execute(
+        sa.text("SELECT 1 FROM pg_indexes WHERE indexname = :name"),
+        {"name": index_name},
+    )
+    return result.scalar() is not None
+
+
+# ---------------------------------------------------------------------------
+# Batch ALTER TYPE helper
+# ---------------------------------------------------------------------------
+
+
+def batch_alter_type(
+    cache: _SchemaCache,
+    columns: list[tuple[str, str, bool, str | None]],
+    cast_suffix: str,
+    type_fn: Callable[[str], str],
+) -> None:
+    """Group columns by table and issue ONE ``ALTER TABLE`` per table.
+
+    Skips columns that don't exist.  Each tuple is
+    ``(table_name, column_name, nullable, server_default)``.
+    """
+    by_table: dict[str, list[tuple[str, str]]] = defaultdict(list)
+    for table, col, _nullable, _default in columns:
+        if not cache.column_exists(table, col):
+            continue
+        by_table[table].append((col, type_fn(col)))
+
+    bind = op.get_bind()
+    for table, col_types in by_table.items():
+        parts = [
+            f"ALTER COLUMN {col} TYPE {target} USING {col}::{cast_suffix}"
+            for col, target in col_types
+        ]
+        if parts:
+            bind.execute(sa.text(f"ALTER TABLE {table} " + ", ".join(parts)))

alembic/versions/20260308_0348_45b118150a78_usage_stats_retention_set_null_on_.py

@@ -5,137 +5,93 @@ Revises: 2d932114930d
 Create Date: 2026-03-08 03:48:49.622091+00:00
 
 """
-from alembic import op
+
 import sqlalchemy as sa
 
+from alembic import op
+from alembic.helpers import new_cache, replace_fk_if_needed
+
 # revision identifiers, used by Alembic.
-revision = '45b118150a78'
+revision = "45b118150a78"
-down_revision = '2d932114930d'
+down_revision = "2d932114930d"
 branch_labels = None
 depends_on = None
 
-
+_TABLES = ["usage", "stats_user_daily", "stats_daily_api_key"]
-# ---------------------------------------------------------------------------
-# Idempotent helpers
-# ---------------------------------------------------------------------------
-
-def _column_exists(table_name: str, column_name: str) -> bool:
-    bind = op.get_bind()
-    result = bind.execute(
-        sa.text(
-            "SELECT 1 FROM information_schema.columns "
-            "WHERE table_name = :table AND column_name = :col"
-        ),
-        {"table": table_name, "col": column_name},
-    )
-    return result.scalar() is not None
-
-
-def _constraint_exists(table_name: str, constraint_name: str) -> bool:
-    bind = op.get_bind()
-    result = bind.execute(
-        sa.text(
-            "SELECT 1 FROM information_schema.table_constraints "
-            "WHERE table_name = :table AND constraint_name = :name"
-        ),
-        {"table": table_name, "name": constraint_name},
-    )
-    return result.scalar() is not None
-
-
-def _fk_ondelete(table_name: str, constraint_name: str) -> str | None:
-    """Return the ON DELETE action for a foreign key, or None if it doesn't exist."""
-    bind = op.get_bind()
-    result = bind.execute(
-        sa.text(
-            "SELECT rc.delete_rule "
-            "FROM information_schema.referential_constraints rc "
-            "JOIN information_schema.table_constraints tc "
-            "  ON rc.constraint_name = tc.constraint_name "
-            "WHERE tc.table_name = :table AND tc.constraint_name = :name"
-        ),
-        {"table": table_name, "name": constraint_name},
-    )
-    row = result.first()
-    return row[0] if row else None
-
-
-def _replace_fk_if_needed(
-    constraint_name: str,
-    table_name: str,
-    ref_table: str,
-    local_cols: list[str],
-    remote_cols: list[str],
-    desired_ondelete: str,
-) -> None:
-    """Drop and recreate a FK only if the current ON DELETE rule differs."""
-    current = _fk_ondelete(table_name, constraint_name)
-    if current and current.upper() == desired_ondelete.upper():
-        return  # already correct
-    if current:
-        op.drop_constraint(constraint_name, table_name, type_='foreignkey')
-    op.create_foreign_key(
-        constraint_name, table_name, ref_table,
-        local_cols, remote_cols, ondelete=desired_ondelete,
-    )
 
 
 def upgrade() -> None:
+    c = new_cache()
+    c.load_columns(_TABLES)
+    c.load_fk_rules(["stats_user_daily", "stats_daily_api_key"])
+
     # --- Usage: add name snapshot columns ---
-    if not _column_exists('usage', 'username'):
+    if not c.column_exists("usage", "username"):
-        op.add_column('usage', sa.Column('username', sa.String(100), nullable=True,
+        op.add_column(
-                                         comment='用户名快照'))
+            "usage", sa.Column("username", sa.String(100), nullable=True, comment="用户名快照")
-    if not _column_exists('usage', 'api_key_name'):
+        )
-        op.add_column('usage', sa.Column('api_key_name', sa.String(200), nullable=True,
+    if not c.column_exists("usage", "api_key_name"):
-                                         comment='API Key 名称快照'))
+        op.add_column(
+            "usage",
+            sa.Column("api_key_name", sa.String(200), nullable=True, comment="API Key 名称快照"),
+        )
 
     # --- StatsUserDaily: CASCADE -> SET NULL, add username snapshot ---
-    _replace_fk_if_needed(
+    replace_fk_if_needed(
-        'stats_user_daily_user_id_fkey', 'stats_user_daily',
+        c,
-        'users', ['user_id'], ['id'], 'SET NULL',
+        "stats_user_daily_user_id_fkey",
+        "stats_user_daily",
+        "users",
+        ["user_id"],
+        ["id"],
+        "SET NULL",
     )
-    op.alter_column('stats_user_daily', 'user_id', existing_type=sa.String(36), nullable=True)
+    op.alter_column("stats_user_daily", "user_id", existing_type=sa.String(36), nullable=True)
-    if not _column_exists('stats_user_daily', 'username'):
+    if not c.column_exists("stats_user_daily", "username"):
-        op.add_column('stats_user_daily', sa.Column('username', sa.String(100), nullable=True,
+        op.add_column(
-                                                    comment='用户名快照（删除用户后仍可追溯）'))
+            "stats_user_daily",
+            sa.Column(
+                "username",
+                sa.String(100),
+                nullable=True,
+                comment="用户名快照（删除用户后仍可追溯）",
+            ),
+        )
 
     # --- StatsDailyApiKey: CASCADE -> SET NULL, add api_key_name snapshot ---
-    _replace_fk_if_needed(
+    replace_fk_if_needed(
-        'stats_daily_api_key_api_key_id_fkey', 'stats_daily_api_key',
+        c,
-        'api_keys', ['api_key_id'], ['id'], 'SET NULL',
+        "stats_daily_api_key_api_key_id_fkey",
+        "stats_daily_api_key",
+        "api_keys",
+        ["api_key_id"],
+        ["id"],
+        "SET NULL",
     )
-    op.alter_column('stats_daily_api_key', 'api_key_id', existing_type=sa.String(36),
+    op.alter_column("stats_daily_api_key", "api_key_id", existing_type=sa.String(36), nullable=True)
-                    nullable=True)
+    if not c.column_exists("stats_daily_api_key", "api_key_name"):
-    if not _column_exists('stats_daily_api_key', 'api_key_name'):
+        op.add_column(
-        op.add_column('stats_daily_api_key', sa.Column('api_key_name', sa.String(200),
+            "stats_daily_api_key",
-                                                       nullable=True,
+            sa.Column(
-                                                       comment='API Key 名称快照（删除 Key 后仍可追溯）'))
+                "api_key_name",
+                sa.String(200),
+                nullable=True,
+                comment="API Key 名称快照（删除 Key 后仍可追溯）",
+            ),
+        )
 
     # --- Backfill: populate snapshots from existing FK joins ---
-    # Single UPDATE per table using LEFT JOINs to fill both columns in one pass.
+    # One LEFT JOIN UPDATE per table covers all rows regardless of which FK is present.
-    # (WHERE ... IS NULL makes these inherently idempotent)
     op.execute("""
         UPDATE usage u
         SET username     = COALESCE(u.username, usr.username),
             api_key_name = COALESCE(u.api_key_name, ak.name)
-        FROM users usr, api_keys ak
+        FROM usage u2
-        WHERE usr.id = u.user_id
+        LEFT JOIN users usr ON usr.id = u2.user_id
-          AND ak.id = u.api_key_id
+        LEFT JOIN api_keys ak ON ak.id = u2.api_key_id
+        WHERE u.id = u2.id
           AND (u.username IS NULL OR u.api_key_name IS NULL)
     """)
-    # Catch rows that have user_id but no api_key_id (or vice versa)
-    op.execute("""
-        UPDATE usage u
-        SET username = usr.username
-        FROM users usr
-        WHERE u.user_id = usr.id AND u.username IS NULL
-    """)
-    op.execute("""
-        UPDATE usage u
-        SET api_key_name = ak.name
-        FROM api_keys ak
-        WHERE u.api_key_id = ak.id AND u.api_key_name IS NULL
-    """)
     op.execute("""
         UPDATE stats_user_daily s
         SET username = usr.username
@@ -151,27 +107,42 @@ def upgrade() -> None:
 
 
 def downgrade() -> None:
+    c = new_cache()
+    c.load_columns(["stats_daily_api_key", "stats_user_daily", "usage"])
+    c.load_fk_rules(["stats_daily_api_key", "stats_user_daily"])
+
     # --- Remove snapshot columns ---
-    if _column_exists('stats_daily_api_key', 'api_key_name'):
+    if c.column_exists("stats_daily_api_key", "api_key_name"):
-        op.drop_column('stats_daily_api_key', 'api_key_name')
+        op.drop_column("stats_daily_api_key", "api_key_name")
-    if _column_exists('stats_user_daily', 'username'):
+    if c.column_exists("stats_user_daily", "username"):
-        op.drop_column('stats_user_daily', 'username')
+        op.drop_column("stats_user_daily", "username")
-    if _column_exists('usage', 'api_key_name'):
+    if c.column_exists("usage", "api_key_name"):
-        op.drop_column('usage', 'api_key_name')
+        op.drop_column("usage", "api_key_name")
-    if _column_exists('usage', 'username'):
+    if c.column_exists("usage", "username"):
-        op.drop_column('usage', 'username')
+        op.drop_column("usage", "username")
 
     # --- StatsDailyApiKey: SET NULL -> CASCADE ---
-    _replace_fk_if_needed(
+    replace_fk_if_needed(
-        'stats_daily_api_key_api_key_id_fkey', 'stats_daily_api_key',
+        c,
-        'api_keys', ['api_key_id'], ['id'], 'CASCADE',
+        "stats_daily_api_key_api_key_id_fkey",
+        "stats_daily_api_key",
+        "api_keys",
+        ["api_key_id"],
+        ["id"],
+        "CASCADE",
+    )
+    op.alter_column(
+        "stats_daily_api_key", "api_key_id", existing_type=sa.String(36), nullable=False
     )
-    op.alter_column('stats_daily_api_key', 'api_key_id', existing_type=sa.String(36),
-                    nullable=False)
 
     # --- StatsUserDaily: SET NULL -> CASCADE ---
-    _replace_fk_if_needed(
+    replace_fk_if_needed(
-        'stats_user_daily_user_id_fkey', 'stats_user_daily',
+        c,
-        'users', ['user_id'], ['id'], 'CASCADE',
+        "stats_user_daily_user_id_fkey",
+        "stats_user_daily",
+        "users",
+        ["user_id"],
+        ["id"],
+        "CASCADE",
     )
-    op.alter_column('stats_user_daily', 'user_id', existing_type=sa.String(36), nullable=False)
+    op.alter_column("stats_user_daily", "user_id", existing_type=sa.String(36), nullable=False)

alembic/versions/20260308_1215_13a4c8f6d9e0_request_candidate_video_task_retention.py

@@ -6,89 +6,32 @@ Create Date: 2026-03-08 12:15:00.000000+00:00
 
 """
 
-from alembic import op
 import sqlalchemy as sa
 
+from alembic import op
+from alembic.helpers import new_cache, replace_fk_if_needed
+
 # revision identifiers, used by Alembic.
 revision = "13a4c8f6d9e0"
 down_revision = "45b118150a78"
 branch_labels = None
 depends_on = None
 
-
+_TABLES = ["request_candidates", "video_tasks"]
-# ---------------------------------------------------------------------------
-# Idempotent helpers
-# ---------------------------------------------------------------------------
-
-def _column_exists(table_name: str, column_name: str) -> bool:
-    bind = op.get_bind()
-    result = bind.execute(
-        sa.text(
-            "SELECT 1 FROM information_schema.columns "
-            "WHERE table_name = :table AND column_name = :col"
-        ),
-        {"table": table_name, "col": column_name},
-    )
-    return result.scalar() is not None
-
-
-def _constraint_exists(table_name: str, constraint_name: str) -> bool:
-    bind = op.get_bind()
-    result = bind.execute(
-        sa.text(
-            "SELECT 1 FROM information_schema.table_constraints "
-            "WHERE table_name = :table AND constraint_name = :name"
-        ),
-        {"table": table_name, "name": constraint_name},
-    )
-    return result.scalar() is not None
-
-
-def _fk_ondelete(table_name: str, constraint_name: str) -> str | None:
-    """Return the ON DELETE action for a foreign key, or None if it doesn't exist."""
-    bind = op.get_bind()
-    result = bind.execute(
-        sa.text(
-            "SELECT rc.delete_rule "
-            "FROM information_schema.referential_constraints rc "
-            "JOIN information_schema.table_constraints tc "
-            "  ON rc.constraint_name = tc.constraint_name "
-            "WHERE tc.table_name = :table AND tc.constraint_name = :name"
-        ),
-        {"table": table_name, "name": constraint_name},
-    )
-    row = result.first()
-    return row[0] if row else None
-
-
-def _replace_fk_if_needed(
-    constraint_name: str,
-    table_name: str,
-    ref_table: str,
-    local_cols: list[str],
-    remote_cols: list[str],
-    desired_ondelete: str,
-) -> None:
-    """Drop and recreate a FK only if the current ON DELETE rule differs."""
-    current = _fk_ondelete(table_name, constraint_name)
-    if current and current.upper() == desired_ondelete.upper():
-        return  # already correct
-    if current:
-        op.drop_constraint(constraint_name, table_name, type_="foreignkey")
-    op.create_foreign_key(
-        constraint_name, table_name, ref_table,
-        local_cols, remote_cols, ondelete=desired_ondelete,
-    )
 
 
 def upgrade() -> None:
+    c = new_cache()
+    c.load_columns(_TABLES)
+    c.load_fk_rules(_TABLES)
+
     # --- request_candidates: add snapshot columns ---
-    if not _column_exists("request_candidates", "username"):
+    if not c.column_exists("request_candidates", "username"):
         op.add_column(
             "request_candidates",
             sa.Column("username", sa.String(length=100), nullable=True, comment="用户名快照"),
         )
-    if not _column_exists("request_candidates", "api_key_name"):
+    if not c.column_exists("request_candidates", "api_key_name"):
         op.add_column(
             "request_candidates",
             sa.Column(
@@ -100,22 +43,32 @@ def upgrade() -> None:
         )
 
     # --- request_candidates: CASCADE -> SET NULL ---
-    _replace_fk_if_needed(
+    replace_fk_if_needed(
-        "request_candidates_user_id_fkey", "request_candidates",
+        c,
-        "users", ["user_id"], ["id"], "SET NULL",
+        "request_candidates_user_id_fkey",
+        "request_candidates",
+        "users",
+        ["user_id"],
+        ["id"],
+        "SET NULL",
     )
-    _replace_fk_if_needed(
+    replace_fk_if_needed(
-        "request_candidates_api_key_id_fkey", "request_candidates",
+        c,
-        "api_keys", ["api_key_id"], ["id"], "SET NULL",
+        "request_candidates_api_key_id_fkey",
+        "request_candidates",
+        "api_keys",
+        ["api_key_id"],
+        ["id"],
+        "SET NULL",
     )
 
     # --- video_tasks: add snapshot columns ---
-    if not _column_exists("video_tasks", "username"):
+    if not c.column_exists("video_tasks", "username"):
         op.add_column(
             "video_tasks",
             sa.Column("username", sa.String(length=100), nullable=True, comment="用户名快照"),
         )
-    if not _column_exists("video_tasks", "api_key_name"):
+    if not c.column_exists("video_tasks", "api_key_name"):
         op.add_column(
             "video_tasks",
             sa.Column(
@@ -128,106 +81,100 @@ def upgrade() -> None:
 
     # --- video_tasks: CASCADE -> SET NULL, user_id nullable ---
     op.alter_column("video_tasks", "user_id", existing_type=sa.String(length=36), nullable=True)
-    _replace_fk_if_needed(
+    replace_fk_if_needed(
-        "video_tasks_user_id_fkey", "video_tasks",
+        c,
-        "users", ["user_id"], ["id"], "SET NULL",
+        "video_tasks_user_id_fkey",
+        "video_tasks",
+        "users",
+        ["user_id"],
+        ["id"],
+        "SET NULL",
     )
-    _replace_fk_if_needed(
+    replace_fk_if_needed(
-        "video_tasks_api_key_id_fkey", "video_tasks",
+        c,
-        "api_keys", ["api_key_id"], ["id"], "SET NULL",
+        "video_tasks_api_key_id_fkey",
+        "video_tasks",
+        "api_keys",
+        ["api_key_id"],
+        ["id"],
+        "SET NULL",
     )
 
     # --- Backfill: populate snapshots from existing FK joins ---
-    # Single UPDATE per table using JOINs to fill both columns in one pass.
+    # One LEFT JOIN UPDATE per table covers all rows regardless of which FK is present.
-    # (WHERE ... IS NULL makes these inherently idempotent)
+    op.execute("""
-
-    # request_candidates: fill both columns where both FKs exist
-    op.execute(
-        """
         UPDATE request_candidates rc
         SET username     = COALESCE(rc.username, usr.username),
             api_key_name = COALESCE(rc.api_key_name, ak.name)
-        FROM users usr, api_keys ak
+        FROM request_candidates rc2
-        WHERE usr.id = rc.user_id
+        LEFT JOIN users usr ON usr.id = rc2.user_id
-          AND ak.id = rc.api_key_id
+        LEFT JOIN api_keys ak ON ak.id = rc2.api_key_id
+        WHERE rc.id = rc2.id
           AND (rc.username IS NULL OR rc.api_key_name IS NULL)
-        """
+        """)
-    )
-    # Catch rows with only user_id or only api_key_id
-    op.execute(
-        """
-        UPDATE request_candidates rc
-        SET username = usr.username
-        FROM users usr
-        WHERE usr.id = rc.user_id AND rc.username IS NULL
-        """
-    )
-    op.execute(
-        """
-        UPDATE request_candidates rc
-        SET api_key_name = ak.name
-        FROM api_keys ak
-        WHERE ak.id = rc.api_key_id AND rc.api_key_name IS NULL
-        """
-    )
 
-    # video_tasks: fill both columns where both FKs exist
+    op.execute("""
-    op.execute(
-        """
         UPDATE video_tasks vt
         SET username     = COALESCE(vt.username, usr.username),
             api_key_name = COALESCE(vt.api_key_name, ak.name)
-        FROM users usr, api_keys ak
+        FROM video_tasks vt2
-        WHERE usr.id = vt.user_id
+        LEFT JOIN users usr ON usr.id = vt2.user_id
-          AND ak.id = vt.api_key_id
+        LEFT JOIN api_keys ak ON ak.id = vt2.api_key_id
+        WHERE vt.id = vt2.id
           AND (vt.username IS NULL OR vt.api_key_name IS NULL)
-        """
+        """)
-    )
-    # Catch rows with only user_id or only api_key_id
-    op.execute(
-        """
-        UPDATE video_tasks vt
-        SET username = usr.username
-        FROM users usr
-        WHERE usr.id = vt.user_id AND vt.username IS NULL
-        """
-    )
-    op.execute(
-        """
-        UPDATE video_tasks vt
-        SET api_key_name = ak.name
-        FROM api_keys ak
-        WHERE ak.id = vt.api_key_id AND vt.api_key_name IS NULL
-        """
-    )
 
 
 def downgrade() -> None:
+    c = new_cache()
+    c.load_columns(_TABLES)
+    c.load_fk_rules(_TABLES)
+
     # --- video_tasks: SET NULL -> default (no action), restore NOT NULL ---
-    _replace_fk_if_needed(
+    replace_fk_if_needed(
-        "video_tasks_api_key_id_fkey", "video_tasks",
+        c,
-        "api_keys", ["api_key_id"], ["id"], "NO ACTION",
+        "video_tasks_api_key_id_fkey",
+        "video_tasks",
+        "api_keys",
+        ["api_key_id"],
+        ["id"],
+        "NO ACTION",
     )
-    _replace_fk_if_needed(
+    replace_fk_if_needed(
-        "video_tasks_user_id_fkey", "video_tasks",
+        c,
-        "users", ["user_id"], ["id"], "NO ACTION",
+        "video_tasks_user_id_fkey",
+        "video_tasks",
+        "users",
+        ["user_id"],
+        ["id"],
+        "NO ACTION",
     )
     op.alter_column("video_tasks", "user_id", existing_type=sa.String(length=36), nullable=False)
-    if _column_exists("video_tasks", "api_key_name"):
+    if c.column_exists("video_tasks", "api_key_name"):
         op.drop_column("video_tasks", "api_key_name")
-    if _column_exists("video_tasks", "username"):
+    if c.column_exists("video_tasks", "username"):
         op.drop_column("video_tasks", "username")
 
     # --- request_candidates: SET NULL -> CASCADE ---
-    _replace_fk_if_needed(
+    replace_fk_if_needed(
-        "request_candidates_api_key_id_fkey", "request_candidates",
+        c,
-        "api_keys", ["api_key_id"], ["id"], "CASCADE",
+        "request_candidates_api_key_id_fkey",
+        "request_candidates",
+        "api_keys",
+        ["api_key_id"],
+        ["id"],
+        "CASCADE",
     )
-    _replace_fk_if_needed(
+    replace_fk_if_needed(
-        "request_candidates_user_id_fkey", "request_candidates",
+        c,
-        "users", ["user_id"], ["id"], "CASCADE",
+        "request_candidates_user_id_fkey",
+        "request_candidates",
+        "users",
+        ["user_id"],
+        ["id"],
+        "CASCADE",
     )
-    if _column_exists("request_candidates", "api_key_name"):
+    if c.column_exists("request_candidates", "api_key_name"):
         op.drop_column("request_candidates", "api_key_name")
-    if _column_exists("request_candidates", "username"):
+    if c.column_exists("request_candidates", "username"):
         op.drop_column("request_candidates", "username")

alembic/versions/20260308_1530_2053ab8ed764_cost_fields_float_to_numeric.py

@@ -6,9 +6,8 @@ Create Date: 2026-03-08 15:30:00.000000+00:00
 
 """
 
-import sqlalchemy as sa
-
 from alembic import op
+from alembic.helpers import batch_alter_type, index_exists, new_cache
 
 # revision identifiers, used by Alembic.
 revision = "2053ab8ed764"
@@ -77,44 +76,7 @@ _COST_COLUMNS: list[tuple[str, str, bool, str | None]] = [
     ("stats_user_daily", "total_cost", False, "0.0"),
 ]
 
-# rate_multiplier uses a smaller precision
+_ALL_TABLES = list({t for t, *_ in _COST_COLUMNS})
-_RATE_MULTIPLIER_TYPE = sa.Numeric(10, 6)
-_COST_TYPE = sa.Numeric(20, 8)
-
-
-def _column_exists(table_name: str, column_name: str) -> bool:
-    bind = op.get_bind()
-    result = bind.execute(
-        sa.text(
-            "SELECT 1 FROM information_schema.columns "
-            "WHERE table_name = :table AND column_name = :col"
-        ),
-        {"table": table_name, "col": column_name},
-    )
-    return result.scalar() is not None
-
-
-def _index_exists(index_name: str) -> bool:
-    bind = op.get_bind()
-    result = bind.execute(
-        sa.text("SELECT 1 FROM pg_indexes WHERE indexname = :name"),
-        {"name": index_name},
-    )
-    return result.scalar() is not None
-
-
-def _is_numeric_type(table_name: str, column_name: str) -> bool:
-    """Check if a column is already numeric type (not float/double precision)."""
-    bind = op.get_bind()
-    result = bind.execute(
-        sa.text(
-            "SELECT data_type FROM information_schema.columns "
-            "WHERE table_name = :table AND column_name = :col"
-        ),
-        {"table": table_name, "col": column_name},
-    )
-    data_type = result.scalar()
-    return data_type == "numeric"
 
 
 def _type_spec(col: str) -> str:
@@ -122,50 +84,20 @@ def _type_spec(col: str) -> str:
     return "NUMERIC(10,6)" if col == "rate_multiplier" else "NUMERIC(20,8)"
 
 
-def _batch_alter_type(
-    columns: list[tuple[str, str, bool, str | None]],
-    cast_suffix: str,
-    type_fn=None,
-) -> None:
-    """Group columns by table and issue ONE ALTER TABLE per table.
-
-    This avoids rewriting the same table N times (once per column).
-    """
-    from collections import defaultdict
-
-    by_table: dict[str, list[tuple[str, str, bool, str | None]]] = defaultdict(list)
-    for table, col, nullable, default in columns:
-        if not _column_exists(table, col):
-            continue
-        by_table[table].append((table, col, nullable, default))
-
-    bind = op.get_bind()
-    for table, cols in by_table.items():
-        # Build a single ALTER TABLE with multiple ALTER COLUMN clauses
-        parts: list[str] = []
-        for _t, col, _nullable, _default in cols:
-            target_type = type_fn(col) if type_fn else _type_spec(col)
-            parts.append(
-                f"ALTER COLUMN {col} TYPE {target_type} USING {col}::{cast_suffix}"
-            )
-        if not parts:
-            continue
-        sql = f"ALTER TABLE {table} " + ", ".join(parts)
-        bind.execute(sa.text(sql))
-
-
 def upgrade() -> None:
+    c = new_cache()
+    c.load_columns(_ALL_TABLES)
+
     # -- 1. cost fields: Float -> Numeric  (batched per table)
-    # Filter out columns that are already numeric
     cols_to_convert = [
-        (t, c, n, d)
+        (t, col, n, d)
-        for t, c, n, d in _COST_COLUMNS
+        for t, col, n, d in _COST_COLUMNS
-        if _column_exists(t, c) and not _is_numeric_type(t, c)
+        if c.column_exists(t, col) and not c.is_numeric(t, col)
     ]
-    _batch_alter_type(cols_to_convert, cast_suffix="numeric", type_fn=_type_spec)
+    batch_alter_type(c, cols_to_convert, cast_suffix="numeric", type_fn=_type_spec)
 
     # -- 2. provider_api_keys composite index
-    if not _index_exists("idx_provider_api_keys_provider_active"):
+    if not index_exists("idx_provider_api_keys_provider_active"):
         op.create_index(
             "idx_provider_api_keys_provider_active",
             "provider_api_keys",
@@ -175,19 +107,23 @@ def upgrade() -> None:
 
 def downgrade() -> None:
     # -- 2. drop composite index
-    if _index_exists("idx_provider_api_keys_provider_active"):
+    if index_exists("idx_provider_api_keys_provider_active"):
         op.drop_index(
             "idx_provider_api_keys_provider_active",
             table_name="provider_api_keys",
         )
 
     # -- 1. Numeric -> Float  (batched per table)
+    c = new_cache()
+    c.load_columns(_ALL_TABLES)
+
     cols_to_revert = [
-        (t, c, n, d)
+        (t, col, n, d)
-        for t, c, n, d in _COST_COLUMNS
+        for t, col, n, d in _COST_COLUMNS
-        if _column_exists(t, c) and _is_numeric_type(t, c)
+        if c.column_exists(t, col) and c.is_numeric(t, col)
     ]
-    _batch_alter_type(
+    batch_alter_type(
+        c,
         cols_to_revert,
         cast_suffix="double precision",
         type_fn=lambda _col: "DOUBLE PRECISION",

alembic/versions/20260309_0100_d7649c1f8e21_video_task_key_id_set_null_on_delete.py

@@ -6,9 +6,7 @@ Create Date: 2026-03-09 01:00:00.000000+00:00
 
 """
 
-import sqlalchemy as sa
+from alembic.helpers import new_cache, replace_fk_if_needed
-
-from alembic import op
 
 # revision identifiers, used by Alembic.
 revision = "d7649c1f8e21"
@@ -20,47 +18,11 @@ _TABLE = "video_tasks"
 _FK_NAME = "video_tasks_key_id_fkey"
 
 
-def _fk_ondelete(table_name: str, constraint_name: str) -> str | None:
-    bind = op.get_bind()
-    result = bind.execute(
-        sa.text(
-            "SELECT rc.delete_rule "
-            "FROM information_schema.referential_constraints rc "
-            "JOIN information_schema.table_constraints tc "
-            "  ON rc.constraint_name = tc.constraint_name "
-            "WHERE tc.table_name = :table AND tc.constraint_name = :name"
-        ),
-        {"table": table_name, "name": constraint_name},
-    )
-    row = result.first()
-    return row[0] if row else None
-
-
-def _replace_fk_if_needed(
-    constraint_name: str,
-    table_name: str,
-    ref_table: str,
-    local_cols: list[str],
-    remote_cols: list[str],
-    desired_ondelete: str,
-) -> None:
-    current = _fk_ondelete(table_name, constraint_name)
-    if current and current.upper() == desired_ondelete.upper():
-        return
-    if current:
-        op.drop_constraint(constraint_name, table_name, type_="foreignkey")
-    op.create_foreign_key(
-        constraint_name,
-        table_name,
-        ref_table,
-        local_cols,
-        remote_cols,
-        ondelete=desired_ondelete,
-    )
-
-
 def upgrade() -> None:
-    _replace_fk_if_needed(
+    c = new_cache()
+    c.load_fk_rules([_TABLE])
+    replace_fk_if_needed(
+        c,
         _FK_NAME,
         _TABLE,
         "provider_api_keys",
@@ -71,7 +33,10 @@ def upgrade() -> None:
 
 
 def downgrade() -> None:
-    _replace_fk_if_needed(
+    c = new_cache()
+    c.load_fk_rules([_TABLE])
+    replace_fk_if_needed(
+        c,
         _FK_NAME,
         _TABLE,
         "provider_api_keys",

src/api/handlers/base/base_handler.py

@@ -533,18 +533,21 @@ class BaseMessageHandler:
 
         target_request_id = request_id or self.request_id
 
+        def _sync_update() -> None:
+            db_gen = get_db()
+            db = next(db_gen)
+            try:
+                UsageService.update_usage_status(
+                    db=db,
+                    request_id=target_request_id,
+                    status="streaming",
+                )
+            finally:
+                db.close()
+
         async def _do_update() -> None:
             try:
-                db_gen = get_db()
+                await asyncio.to_thread(_sync_update)
-                db = next(db_gen)
-                try:
-                    UsageService.update_usage_status(
-                        db=db,
-                        request_id=target_request_id,
-                        status="streaming",
-                    )
-                finally:
-                    db.close()
             except Exception as e:
                 logger.warning(f"[{target_request_id}] 更新 Usage 状态为 streaming 失败: {e}")
 
@@ -585,29 +588,36 @@ class BaseMessageHandler:
                 f"ctx.provider_name={ctx.provider_name}, ctx.provider_id={ctx.provider_id}"
             )
 
+        # Capture mutable ctx attrs before handing off to thread
+        provider_request_headers = ctx.provider_request_headers or None
+        provider_request_body = ctx.provider_request_body
+
+        def _sync_update() -> None:
+            db_gen = get_db()
+            db = next(db_gen)
+            try:
+                UsageService.update_usage_status(
+                    db=db,
+                    request_id=target_request_id,
+                    status="streaming",
+                    provider=provider,
+                    target_model=target_model,
+                    provider_id=provider_id,
+                    provider_endpoint_id=endpoint_id,
+                    provider_api_key_id=key_id,
+                    first_byte_time_ms=first_byte_time_ms,
+                    api_format=api_format,
+                    endpoint_api_format=endpoint_api_format,
+                    has_format_conversion=has_format_conversion,
+                    provider_request_headers=provider_request_headers,
+                    provider_request_body=provider_request_body,
+                )
+            finally:
+                db.close()
+
         async def _do_update() -> None:
             try:
-                db_gen = get_db()
+                await asyncio.to_thread(_sync_update)
-                db = next(db_gen)
-                try:
-                    UsageService.update_usage_status(
-                        db=db,
-                        request_id=target_request_id,
-                        status="streaming",
-                        provider=provider,
-                        target_model=target_model,
-                        provider_id=provider_id,
-                        provider_endpoint_id=endpoint_id,
-                        provider_api_key_id=key_id,
-                        first_byte_time_ms=first_byte_time_ms,
-                        api_format=api_format,
-                        endpoint_api_format=endpoint_api_format,
-                        has_format_conversion=has_format_conversion,
-                        provider_request_headers=ctx.provider_request_headers or None,
-                        provider_request_body=ctx.provider_request_body,
-                    )
-                finally:
-                    db.close()
             except Exception as e:
                 logger.warning(f"[{target_request_id}] 更新 Usage 状态为 streaming 失败: {e}")
 

src/api/handlers/base/stream_telemetry.py

@@ -392,56 +392,67 @@ class StreamTelemetryRecorder:
         if not ctx.attempt_id:
             return
 
-        from src.services.request.candidate import RequestCandidateService
+        # Capture all needed ctx attrs before handing off to thread
+        attempt_id = ctx.attempt_id
+        is_success = ctx.is_success()
+        is_disconnected = ctx.is_client_disconnected()
+        status_code = ctx.status_code
+        data_count = ctx.data_count
+        rectified = ctx.rectified
+        proxy_info = ctx.proxy_info
+        first_byte_time_ms_val = ctx.first_byte_time_ms
+        upstream_response = ctx.upstream_response
+        error_message = ctx.error_message
 
-        extra_data: dict[str, Any] = {
+        def _sync() -> None:
-            "stream_completed": ctx.is_success(),
+            from src.services.request.candidate import RequestCandidateService
-            "data_count": ctx.data_count,
-        }
-        if ctx.rectified:
-            extra_data["rectified"] = True
-        if ctx.proxy_info:
-            extra_data["proxy"] = ctx.proxy_info
-        if ctx.first_byte_time_ms is not None:
-            # 计算候选自身的 TTFB
-            first_byte_time_ms = RequestCandidateService.calculate_candidate_ttfb(
-                db=db,
-                candidate_id=ctx.attempt_id,
-                request_start_time=request_start_time,
-                global_first_byte_time_ms=ctx.first_byte_time_ms,
-            )
-            extra_data["first_byte_time_ms"] = first_byte_time_ms
 
-        if ctx.is_success():
+            extra_data: dict[str, Any] = {
-            RequestCandidateService.mark_candidate_success(
+                "stream_completed": is_success,
-                db=db,
+                "data_count": data_count,
-                candidate_id=ctx.attempt_id,
+            }
-                status_code=ctx.status_code,
+            if rectified:
-                latency_ms=response_time_ms,
+                extra_data["rectified"] = True
-                extra_data=extra_data,
+            if proxy_info:
-            )
+                extra_data["proxy"] = proxy_info
-        elif ctx.is_client_disconnected():
+            if first_byte_time_ms_val is not None:
-            RequestCandidateService.mark_candidate_cancelled(
+                candidate_ttfb = RequestCandidateService.calculate_candidate_ttfb(
-                db=db,
+                    db=db,
-                candidate_id=ctx.attempt_id,
+                    candidate_id=attempt_id,
-                status_code=ctx.status_code,
+                    request_start_time=request_start_time,
-                latency_ms=response_time_ms,
+                    global_first_byte_time_ms=first_byte_time_ms_val,
-                extra_data=extra_data,
+                )
-            )
+                extra_data["first_byte_time_ms"] = candidate_ttfb
-        else:
+
-            # 请求链路追踪使用 upstream_response（原始响应），回退到 error_message（友好消息）
+            if is_success:
-            trace_error_message = (
+                RequestCandidateService.mark_candidate_success(
-                ctx.upstream_response or ctx.error_message or f"HTTP {ctx.status_code}"
+                    db=db,
-            )
+                    candidate_id=attempt_id,
-            RequestCandidateService.mark_candidate_failed(
+                    status_code=status_code,
-                db=db,
+                    latency_ms=response_time_ms,
-                candidate_id=ctx.attempt_id,
+                    extra_data=extra_data,
-                error_type="stream_error",
+                )
-                error_message=trace_error_message,
+            elif is_disconnected:
-                status_code=ctx.status_code,
+                RequestCandidateService.mark_candidate_cancelled(
-                latency_ms=response_time_ms,
+                    db=db,
-                extra_data=extra_data,
+                    candidate_id=attempt_id,
-            )
+                    status_code=status_code,
+                    latency_ms=response_time_ms,
+                    extra_data=extra_data,
+                )
+            else:
+                trace_error_message = upstream_response or error_message or f"HTTP {status_code}"
+                RequestCandidateService.mark_candidate_failed(
+                    db=db,
+                    candidate_id=attempt_id,
+                    error_type="stream_error",
+                    error_message=trace_error_message,
+                    status_code=status_code,
+                    latency_ms=response_time_ms,
+                    extra_data=extra_data,
+                )
+
+        await asyncio.to_thread(_sync)
 
     async def _update_usage_status_on_error(
         self,
@@ -474,10 +485,12 @@ class StreamTelemetryRecorder:
         error_message: str | None = None,
     ) -> None:
         """直接更新 Usage 表的状态字段"""
-        try:
+        request_id = self.request_id
+
+        def _sync() -> None:
             from src.models.database import Usage
 
-            usage = db.query(Usage).filter(Usage.request_id == self.request_id).first()
+            usage = db.query(Usage).filter(Usage.request_id == request_id).first()
             if usage:
                 setattr(usage, "status", status)
                 setattr(usage, "status_code", status_code)
@@ -485,7 +498,10 @@ class StreamTelemetryRecorder:
                 if error_message:
                     setattr(usage, "error_message", error_message)
                 db.commit()
-                logger.debug(f"[{self.request_id}] Usage 状态已更新: {status}")
+                logger.debug(f"[{request_id}] Usage 状态已更新: {status}")
+
+        try:
+            await asyncio.to_thread(_sync)
         except Exception as e:
             logger.error(f"[{self.request_id}] 直接更新 Usage 状态失败: {e}")
 

src/services/candidate/failover.py

@@ -24,6 +24,17 @@ from .policy import FailoverAction, RetryMode, RetryPolicy, SkipPolicy
 from .recorder import CandidateRecorder
 from .schema import CandidateKey
 
+_DISCONNECT_EXCEPTION_NAMES = frozenset({"ClientDisconnectedException"})
+
+
+def _is_client_disconnected(exc: Exception) -> bool:
+    """检查异常(或其 cause)是否为客户端断连，避免循环 import。"""
+    for obj in (exc, getattr(exc, "cause", None)):
+        if obj is not None and type(obj).__name__ in _DISCONNECT_EXCEPTION_NAMES:
+            return True
+    return False
+
+
 _SENSITIVE_PATTERN = re.compile(
     r"(api[_-]?key|token|bearer|authorization)[=:\s]+\S+",
     re.IGNORECASE,
@@ -66,6 +77,14 @@ class FailoverEngine:
         self._error_classifier = error_classifier or ErrorClassifier(db=db)
         self._recorder = recorder or CandidateRecorder(db)
 
+    async def _db_op(self, func: Callable[[], Any]) -> Any:
+        """将同步 DB 操作放到线程池执行，避免阻塞 asyncio 事件循环。
+
+        当事件循环被同步 db.commit() / db.execute() 阻塞时，
+        Hub PING 心跳无法发送，导致 worker idle timeout 断连，整个服务不可用。
+        """
+        return await asyncio.to_thread(func)
+
     @staticmethod
     def _collect_error_messages(error: Exception | None) -> str:
         if error is None:
@@ -192,7 +211,7 @@ class FailoverEngine:
         except Exception:
             return False
 
-    def _mark_remaining_cancelled(
+    async def _mark_remaining_cancelled(
         self,
         *,
         candidate_record_map: dict[tuple[int, int], str] | None,
@@ -204,8 +223,8 @@ class FailoverEngine:
         if not candidate_record_map:
             return
 
-        now = datetime.now(timezone.utc)
+        # Pre-compute outside thread (reads ProviderCandidate attrs that may not be thread-safe)
-        updated = False
+        record_ids: list[str] = []
         for candidate_idx, cand in enumerate(candidates):
             if candidate_idx < from_candidate_idx:
                 continue
@@ -214,11 +233,18 @@ class FailoverEngine:
                 if candidate_idx == from_candidate_idx and retry_idx < from_retry_idx:
                     continue
                 record_id = candidate_record_map.get((candidate_idx, retry_idx))
-                if not record_id:
+                if record_id:
-                    continue
+                    record_ids.append(record_id)
+
+        if not record_ids:
+            return
+
+        def _do() -> None:
+            now = datetime.now(timezone.utc)
+            for rid in record_ids:
                 self.db.execute(
                     update(RequestCandidate)
-                    .where(RequestCandidate.id == record_id)
+                    .where(RequestCandidate.id == rid)
                     .where(RequestCandidate.status.in_(["available", "pending"]))
                     .values(
                         status="cancelled",
@@ -227,11 +253,10 @@ class FailoverEngine:
                         finished_at=now,
                     )
                 )
-                updated = True
-
-        if updated:
             self.db.commit()
 
+        await self._db_op(_do)
+
     def _append_cancelled_fallback_candidate_keys(
         self,
         *,
@@ -295,7 +320,7 @@ class FailoverEngine:
             from_candidate_idx,
             from_retry_idx,
         )
-        self._mark_remaining_cancelled(
+        await self._mark_remaining_cancelled(
             candidate_record_map=candidate_record_map,
             candidates=candidates,
             from_candidate_idx=from_candidate_idx,
@@ -322,6 +347,46 @@ class FailoverEngine:
             attempt_count=attempt_count,
         )
 
+    async def _build_disconnected_result(
+        self,
+        *,
+        candidate_record_map: dict[tuple[int, int], str] | None,
+        candidate_keys_fallback: list[CandidateKey],
+        candidates: list[ProviderCandidate],
+        from_candidate_idx: int,
+        from_retry_idx: int,
+        retry_policy: RetryPolicy,
+        request_id: str | None,
+        attempt_count: int,
+    ) -> ExecutionResult:
+        """客户端已断连，立即终止故障转移并返回结果。"""
+        await self._mark_remaining_cancelled(
+            candidate_record_map=candidate_record_map,
+            candidates=candidates,
+            from_candidate_idx=from_candidate_idx,
+            from_retry_idx=from_retry_idx,
+            retry_policy=retry_policy,
+        )
+        self._append_cancelled_fallback_candidate_keys(
+            fallback=candidate_keys_fallback,
+            candidates=candidates,
+            from_candidate_idx=from_candidate_idx,
+            from_retry_idx=from_retry_idx,
+            retry_policy=retry_policy,
+        )
+        return ExecutionResult(
+            success=False,
+            error_type="ClientDisconnected",
+            error_message="client_disconnected",
+            last_status_code=499,
+            candidate_keys=self._get_candidate_keys(
+                request_id=request_id,
+                fallback=candidate_keys_fallback,
+                candidates=candidates,
+            ),
+            attempt_count=attempt_count,
+        )
+
     async def execute(
         self,
         *,
@@ -391,7 +456,7 @@ class FailoverEngine:
             if should_skip:
                 # PRE_EXPAND: mark all retry slots skipped.
                 if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
-                    self._mark_candidate_skipped(
+                    await self._mark_candidate_skipped(
                         candidate_record_map=candidate_record_map,
                         candidate_index=candidate_index,
                         candidate=candidate,
@@ -490,17 +555,9 @@ class FailoverEngine:
                     candidate, candidate_index, retry_index, record_id, attempt_count, max_attempts
                 )
 
-                # Mark pending
+                # Mark pending + commit BEFORE await
-                now = datetime.now(timezone.utc)
+                # (avoid holding DB connections during slow upstream calls)
-                if record_id:
+                await self._mark_pending_and_commit(record_id)
-                    self._update_record(
-                        record_id,
-                        status="pending",
-                        started_at=now,
-                    )
-
-                # Commit BEFORE await (avoid holding DB connections during slow upstream calls)
-                self._commit_before_await()
 
                 try:
                     attempt_result = await self._execute_attempt(
@@ -512,7 +569,7 @@ class FailoverEngine:
 
                     # PRE_EXPAND: mark unused slots after request ends (success)
                     if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
-                        self._mark_remaining_slots_unused(
+                        await self._mark_remaining_slots_unused(
                             candidate_record_map=candidate_record_map,
                             candidates=candidates,
                             success_candidate_idx=candidate_index,
@@ -542,7 +599,7 @@ class FailoverEngine:
 
                 except StreamProbeError as exc:
                     last_status_code = exc.http_status
-                    self._record_attempt_failure(record_id, exc, exc.http_status)
+                    await self._record_attempt_failure(record_id, exc, exc.http_status)
                     action = FailoverAction.CONTINUE
                     consecutive_failures += 1
                     await self._apply_retry_pacing(
@@ -553,6 +610,19 @@ class FailoverEngine:
                     )
 
                 except Exception as exc:
+                    if _is_client_disconnected(exc):
+                        await self._record_attempt_failure(record_id, exc, 499)
+                        return await self._build_disconnected_result(
+                            candidate_record_map=candidate_record_map,
+                            candidate_keys_fallback=candidate_keys_fallback,
+                            candidates=candidates,
+                            from_candidate_idx=candidate_index,
+                            from_retry_idx=retry_index + 1,
+                            retry_policy=retry_policy,
+                            request_id=request_id,
+                            attempt_count=attempt_count,
+                        )
+
                     outcome = await self._handle_attempt_error(
                         exc,
                         candidate=candidate,
@@ -587,7 +657,7 @@ class FailoverEngine:
                 if action == FailoverAction.CONTINUE:
                     # PRE_EXPAND: if we break early, mark remaining retries of this candidate unused.
                     if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
-                        self._mark_candidate_remaining_retries_unused(
+                        await self._mark_candidate_remaining_retries_unused(
                             candidate_record_map=candidate_record_map,
                             candidate_idx=candidate_index,
                             from_retry_idx=retry_index + 1,
@@ -603,7 +673,7 @@ class FailoverEngine:
 
         # exhausted: PRE_EXPAND should not leave 'available' records behind
         if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
-            self._mark_all_remaining_available_unused(candidate_record_map)
+            await self._mark_all_remaining_available_unused(candidate_record_map)
 
         return ExecutionResult(
             success=False,
@@ -670,7 +740,7 @@ class FailoverEngine:
                     or "pool_skipped"
                 )
                 if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
-                    self._mark_retry_indices_status(
+                    await self._mark_retry_indices_status(
                         candidate_record_map=candidate_record_map,
                         candidate_idx=candidate_index,
                         retry_indices=range(
@@ -749,15 +819,7 @@ class FailoverEngine:
                     max_attempts,
                 )
 
-                now = datetime.now(timezone.utc)
+                await self._mark_pending_and_commit(record_id)
-                if record_id:
-                    self._update_record(
-                        record_id,
-                        status="pending",
-                        started_at=now,
-                    )
-
-                self._commit_before_await()
 
                 try:
                     attempt_result = await self._execute_attempt(
@@ -768,7 +830,7 @@ class FailoverEngine:
                     last_status_code = int(getattr(attempt_result, "http_status", 0) or 0)
 
                     if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
-                        self._mark_remaining_slots_unused(
+                        await self._mark_remaining_slots_unused(
                             candidate_record_map=candidate_record_map,
                             candidates=candidates,
                             success_candidate_idx=candidate_index,
@@ -803,7 +865,7 @@ class FailoverEngine:
 
                 except StreamProbeError as exc:
                     last_status_code = exc.http_status
-                    self._record_attempt_failure(record_id, exc, exc.http_status)
+                    await self._record_attempt_failure(record_id, exc, exc.http_status)
                     action = FailoverAction.CONTINUE
                     consecutive_failures += 1
                     await self._apply_retry_pacing(
@@ -814,6 +876,24 @@ class FailoverEngine:
                     )
 
                 except Exception as exc:
+                    if _is_client_disconnected(exc):
+                        await self._record_attempt_failure(record_id, exc, 499)
+                        return (
+                            await self._build_disconnected_result(
+                                candidate_record_map=candidate_record_map,
+                                candidate_keys_fallback=candidate_keys_fallback,
+                                candidates=candidates,
+                                from_candidate_idx=candidate_index,
+                                from_retry_idx=composite_retry_index + 1,
+                                retry_policy=retry_policy,
+                                request_id=request_id,
+                                attempt_count=attempt_count,
+                            ),
+                            attempt_count,
+                            consecutive_failures,
+                            499,
+                        )
+
                     outcome = await self._handle_pool_attempt_error(
                         exc,
                         candidate=candidate,
@@ -843,7 +923,7 @@ class FailoverEngine:
                         # max_retries_for_key may have been shrunk by error handler;
                         # mark unused up to the *original* retry_slots_per_key to cover
                         # all pre-created records.
-                        self._mark_retry_indices_status(
+                        await self._mark_retry_indices_status(
                             candidate_record_map=candidate_record_map,
                             candidate_idx=candidate_index,
                             retry_indices=range(
@@ -862,7 +942,7 @@ class FailoverEngine:
                 # should not terminate the entire request because other providers may still succeed.
                 # When handler_used=True, TaskService raises directly for true STOP semantics.
                 if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
-                    self._mark_candidate_remaining_retries_unused(
+                    await self._mark_candidate_remaining_retries_unused(
                         candidate_record_map=candidate_record_map,
                         candidate_idx=candidate_index,
                         from_retry_idx=composite_retry_index + 1,
@@ -959,7 +1039,7 @@ class FailoverEngine:
                     candidate, is_success=True, response_text=body_text
                 )
                 if rule_action == FailoverAction.CONTINUE:
-                    self._record_attempt_failure(
+                    await self._record_attempt_failure(
                         record_id,
                         Exception("success_failover_pattern matched"),
                         200,
@@ -969,43 +1049,55 @@ class FailoverEngine:
                         http_status=200,
                     )
 
-        self._record_attempt_success(record_id, attempt_result)
+        await self._record_attempt_success(record_id, attempt_result)
         return attempt_result
 
-    def _record_attempt_success(self, record_id: str | None, attempt_result: AttemptResult) -> None:
+    async def _record_attempt_success(
+        self, record_id: str | None, attempt_result: AttemptResult
+    ) -> None:
         """Mark attempt record as success/streaming."""
         if not record_id:
             return
-        if attempt_result.kind == AttemptKind.STREAM:
-            self._update_record(
-                record_id,
-                status="streaming",
-                status_code=attempt_result.http_status,
-            )
-        else:
-            self._update_record(
-                record_id,
-                status="success",
-                status_code=attempt_result.http_status,
-                finished_at=datetime.now(timezone.utc),
-            )
-        self.db.commit()
 
-    def _record_attempt_failure(
+        def _do() -> None:
+            if attempt_result.kind == AttemptKind.STREAM:
+                self._update_record(
+                    record_id,
+                    status="streaming",
+                    status_code=attempt_result.http_status,
+                )
+            else:
+                self._update_record(
+                    record_id,
+                    status="success",
+                    status_code=attempt_result.http_status,
+                    finished_at=datetime.now(timezone.utc),
+                )
+            self.db.commit()
+
+        await self._db_op(_do)
+
+    async def _record_attempt_failure(
         self, record_id: str | None, exc: Exception, status_code: int | None = None
     ) -> None:
         """Mark attempt record as failed."""
         if not record_id:
             return
-        self._update_record(
+        error_type = type(exc).__name__
-            record_id,
+        error_message = self._sanitize(str(exc))
-            status="failed",
+
-            status_code=status_code,
+        def _do() -> None:
-            error_type=type(exc).__name__,
+            self._update_record(
-            error_message=self._sanitize(str(exc)),
+                record_id,
-            finished_at=datetime.now(timezone.utc),
+                status="failed",
-        )
+                status_code=status_code,
-        self.db.commit()
+                error_type=error_type,
+                error_message=error_message,
+                finished_at=datetime.now(timezone.utc),
+            )
+            self.db.commit()
+
+        await self._db_op(_do)
 
     async def _handle_attempt_error(
         self,
@@ -1046,7 +1138,7 @@ class FailoverEngine:
 
         if outcome.action == FailoverAction.STOP:
             if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
-                self._mark_remaining_slots_unused(
+                await self._mark_remaining_slots_unused(
                     candidate_record_map=candidate_record_map,
                     candidates=candidates,
                     success_candidate_idx=candidate_index,
@@ -1120,7 +1212,7 @@ class FailoverEngine:
             last_status_code = int(getattr(exc, "status_code", 0) or 0) or int(
                 getattr(exc, "http_status", 0) or 0
             )
-            self._record_attempt_failure(record_id, exc, last_status_code or None)
+            await self._record_attempt_failure(record_id, exc, last_status_code or None)
 
         return AttemptErrorOutcome(
             action=action,
@@ -1194,13 +1286,33 @@ class FailoverEngine:
             )
         return result
 
-    def _commit_before_await(self) -> None:
+    async def _commit_before_await(self) -> None:
-        if self.db.in_transaction():
+        def _do() -> None:
-            try:
+            if self.db.in_transaction():
-                self.db.commit()
+                try:
-            except Exception:
+                    self.db.commit()
-                self.db.rollback()
+                except Exception:
-                raise
+                    self.db.rollback()
+                    raise
+
+        await self._db_op(_do)
+
+    async def _mark_pending_and_commit(self, record_id: str | None) -> None:
+        """Mark record as pending and commit, all within a worker thread."""
+
+        def _do() -> None:
+            if record_id:
+                self._update_record(
+                    record_id, status="pending", started_at=datetime.now(timezone.utc)
+                )
+            if self.db.in_transaction():
+                try:
+                    self.db.commit()
+                except Exception:
+                    self.db.rollback()
+                    raise
+
+        await self._db_op(_do)
 
     def _update_record(self, record_id: str, /, **values: Any) -> None:
         self.db.execute(
@@ -1221,23 +1333,31 @@ class FailoverEngine:
     ) -> str:
         # Create "available" record, then caller will mark pending.
         extra = self._build_pool_extra_data(candidate)
-        row = RequestCandidateService.create_candidate(
+        provider_id = str(candidate.provider.id)
-            db=self.db,
+        endpoint_id = str(candidate.endpoint.id)
-            request_id=request_id,
+        key_id = str(candidate.key.id)
-            candidate_index=candidate_index,
+        is_cached = bool(getattr(candidate, "is_cached", False))
-            retry_index=retry_index,
+
-            user_id=user_id,
+        def _do() -> str:
-            api_key_id=api_key_id,
+            row = RequestCandidateService.create_candidate(
-            username=username,
+                db=self.db,
-            api_key_name=api_key_name,
+                request_id=request_id,
-            provider_id=str(candidate.provider.id),
+                candidate_index=candidate_index,
-            endpoint_id=str(candidate.endpoint.id),
+                retry_index=retry_index,
-            key_id=str(candidate.key.id),
+                user_id=user_id,
-            status="available",
+                api_key_id=api_key_id,
-            is_cached=bool(getattr(candidate, "is_cached", False)),
+                username=username,
-            extra_data=extra,
+                api_key_name=api_key_name,
-        )
+                provider_id=provider_id,
-        return str(row.id)
+                endpoint_id=endpoint_id,
+                key_id=key_id,
+                status="available",
+                is_cached=is_cached,
+                extra_data=extra,
+            )
+            return str(row.id)
+
+        return await self._db_op(_do)
 
     async def _create_skipped_record(
         self,
@@ -1253,27 +1373,35 @@ class FailoverEngine:
         skip_reason: str | None,
     ) -> str:
         extra = self._build_pool_extra_data(candidate)
-        row = RequestCandidateService.create_candidate(
+        provider_id = str(candidate.provider.id)
-            db=self.db,
+        endpoint_id = str(candidate.endpoint.id)
-            request_id=request_id,
+        key_id = str(candidate.key.id)
-            candidate_index=candidate_index,
+        is_cached = bool(getattr(candidate, "is_cached", False))
-            retry_index=retry_index,
+
-            user_id=user_id,
+        def _do() -> str:
-            api_key_id=api_key_id,
+            row = RequestCandidateService.create_candidate(
-            username=username,
+                db=self.db,
-            api_key_name=api_key_name,
+                request_id=request_id,
-            provider_id=str(candidate.provider.id),
+                candidate_index=candidate_index,
-            endpoint_id=str(candidate.endpoint.id),
+                retry_index=retry_index,
-            key_id=str(candidate.key.id),
+                user_id=user_id,
-            status="skipped",
+                api_key_id=api_key_id,
-            skip_reason=skip_reason,
+                username=username,
-            is_cached=bool(getattr(candidate, "is_cached", False)),
+                api_key_name=api_key_name,
-            extra_data=extra,
+                provider_id=provider_id,
-        )
+                endpoint_id=endpoint_id,
-        # ensure visible for subsequent recorder reads
+                key_id=key_id,
-        if self.db.in_transaction():
+                status="skipped",
-            self.db.commit()
+                skip_reason=skip_reason,
-        return str(row.id)
+                is_cached=is_cached,
+                extra_data=extra,
+            )
+            # ensure visible for subsequent recorder reads
+            if self.db.in_transaction():
+                self.db.commit()
+            return str(row.id)
+
+        return await self._db_op(_do)
 
     def _build_pool_extra_data(self, candidate: ProviderCandidate) -> dict[str, Any]:
         extra: dict[str, Any] = {}
@@ -1576,7 +1704,7 @@ class FailoverEngine:
                 self._sanitize(str(inner)),
             )
 
-    def _mark_candidate_skipped(
+    async def _mark_candidate_skipped(
         self,
         *,
         candidate_record_map: dict[tuple[int, int], str],
@@ -1586,19 +1714,23 @@ class FailoverEngine:
         skip_reason: str | None,
     ) -> None:
         max_retries = self._get_max_retries(candidate, retry_policy)
-        now = datetime.now(timezone.utc)
+        record_ids = [
-        for retry_index in range(max_retries):
+            candidate_record_map[candidate_index, ri]
-            record_id = candidate_record_map.get((candidate_index, retry_index))
+            for ri in range(max_retries)
-            if record_id:
+            if (candidate_index, ri) in candidate_record_map
-                self._update_record(
+        ]
-                    record_id,
+        if not record_ids:
-                    status="skipped",
+            return
-                    skip_reason=skip_reason,
-                    finished_at=now,
-                )
-        self.db.commit()
 
-    def _mark_remaining_slots_unused(
+        def _do() -> None:
+            now = datetime.now(timezone.utc)
+            for rid in record_ids:
+                self._update_record(rid, status="skipped", skip_reason=skip_reason, finished_at=now)
+            self.db.commit()
+
+        await self._db_op(_do)
+
+    async def _mark_remaining_slots_unused(
         self,
         *,
         candidate_record_map: dict[tuple[int, int], str],
@@ -1607,7 +1739,7 @@ class FailoverEngine:
         success_retry_idx: int,
         retry_policy: RetryPolicy,
     ) -> None:
-        now = datetime.now(timezone.utc)
+        record_ids: list[str] = []
         for candidate_idx, cand in enumerate(candidates):
             max_retries = self._get_max_retries(cand, retry_policy)
             for retry_idx in range(max_retries):
@@ -1617,14 +1749,20 @@ class FailoverEngine:
                     continue
                 record_id = candidate_record_map.get((candidate_idx, retry_idx))
                 if record_id:
-                    self._update_record(
+                    record_ids.append(record_id)
-                        record_id,
-                        status="unused",
-                        finished_at=now,
-                    )
-        self.db.commit()
 
-    def _mark_candidate_remaining_retries_unused(
+        if not record_ids:
+            return
+
+        def _do() -> None:
+            now = datetime.now(timezone.utc)
+            for rid in record_ids:
+                self._update_record(rid, status="unused", finished_at=now)
+            self.db.commit()
+
+        await self._db_op(_do)
+
+    async def _mark_candidate_remaining_retries_unused(
         self,
         *,
         candidate_record_map: dict[tuple[int, int], str],
@@ -1633,21 +1771,27 @@ class FailoverEngine:
         retry_policy: RetryPolicy,
     ) -> None:
         # Only meaningful for PRE_EXPAND.
-        # We don't have access to candidate object list here, so infer max_retries from map keys.
-        # Fallback to retry_policy.max_retries.
-        now = datetime.now(timezone.utc)
-        # try best-effort upper bound
         upper = max(
             (ri for (ci, ri) in candidate_record_map.keys() if ci == candidate_idx),
             default=retry_policy.max_retries - 1,
         )
-        for retry_idx in range(from_retry_idx, upper + 1):
+        record_ids = [
-            record_id = candidate_record_map.get((candidate_idx, retry_idx))
+            candidate_record_map[candidate_idx, ri]
-            if record_id:
+            for ri in range(from_retry_idx, upper + 1)
-                self._update_record(record_id, status="unused", finished_at=now)
+            if (candidate_idx, ri) in candidate_record_map
-        self.db.commit()
+        ]
+        if not record_ids:
+            return
 
-    def _mark_retry_indices_status(
+        def _do() -> None:
+            now = datetime.now(timezone.utc)
+            for rid in record_ids:
+                self._update_record(rid, status="unused", finished_at=now)
+            self.db.commit()
+
+        await self._db_op(_do)
+
+    async def _mark_retry_indices_status(
         self,
         *,
         candidate_record_map: dict[tuple[int, int], str],
@@ -1656,33 +1800,45 @@ class FailoverEngine:
         status: str,
         skip_reason: str | None = None,
     ) -> None:
-        now = datetime.now(timezone.utc)
+        record_ids = [
-        for retry_idx in retry_indices:
+            candidate_record_map[candidate_idx, ri]
-            record_id = candidate_record_map.get((candidate_idx, retry_idx))
+            for ri in retry_indices
-            if not record_id:
+            if (candidate_idx, ri) in candidate_record_map
-                continue
+        ]
-            values: dict[str, Any] = {"status": status, "finished_at": now}
+        if not record_ids:
-            if status == "skipped":
+            return
-                values["skip_reason"] = skip_reason
-            self._update_record(record_id, **values)
-        self.db.commit()
 
-    def _mark_all_remaining_available_unused(
+        def _do() -> None:
+            now = datetime.now(timezone.utc)
+            for rid in record_ids:
+                values: dict[str, Any] = {"status": status, "finished_at": now}
+                if status == "skipped":
+                    values["skip_reason"] = skip_reason
+                self._update_record(rid, **values)
+            self.db.commit()
+
+        await self._db_op(_do)
+
+    async def _mark_all_remaining_available_unused(
         self, candidate_record_map: dict[tuple[int, int], str]
     ) -> None:
         # As a safety net: do not leave available records behind in PRE_EXPAND mode.
-        try:
+        ids = list(candidate_record_map.values())
-            ids = list(candidate_record_map.values())
+        if not ids:
-            if not ids:
+            return
-                return
+
-            now = datetime.now(timezone.utc)
+        def _do() -> None:
-            self.db.execute(
+            try:
-                update(RequestCandidate)
+                now = datetime.now(timezone.utc)
-                .where(RequestCandidate.id.in_(ids))
+                self.db.execute(
-                .where(RequestCandidate.status == "available")
+                    update(RequestCandidate)
-                .values(status="unused", finished_at=now)
+                    .where(RequestCandidate.id.in_(ids))
-            )
+                    .where(RequestCandidate.status == "available")
-            self.db.commit()
+                    .values(status="unused", finished_at=now)
-        except Exception:
+                )
-            self.db.rollback()
+                self.db.commit()
-            raise
+            except Exception:
+                self.db.rollback()
+                raise
+
+        await self._db_op(_do)

src/services/usage/recording.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
+import asyncio
 import uuid
-from decimal import Decimal
 from datetime import datetime, timezone
+from decimal import Decimal
 from typing import Any
 
 from sqlalchemy.orm import Session
@@ -421,102 +422,108 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
         usage_params, total_cost = await cls._prepare_usage_record(params)
         total_cost = to_money_decimal(total_cost)
 
-        # 检查是否已存在相同 request_id 的记录
+        def _sync_record() -> Usage:
-        existing_usage = (
+            """同步 DB 操作: with_for_update + 批量 update + commit"""
-            db.query(Usage).filter(Usage.request_id == request_id).with_for_update().first()
+            from sqlalchemy import func as sql_func
-        )
+            from sqlalchemy import update as sa_update
-        if existing_usage:
+
-            if cls._is_usage_finalized(existing_usage):
+            from src.models.database import ApiKey as ApiKeyModel
-                logger.debug(
+            from src.models.database import GlobalModel
-                    "request_id {} 已完成结算，跳过重复记账 (billing_status={})",
+
-                    request_id,
+            # 检查是否已存在相同 request_id 的记录
-                    getattr(existing_usage, "billing_status", None),
+            existing_usage = (
-                )
+                db.query(Usage).filter(Usage.request_id == request_id).with_for_update().first()
-                return existing_usage
-            logger.debug(
-                f"request_id {request_id} 已存在，更新现有记录 "
-                f"(status: {existing_usage.status} -> {status})"
             )
-            cls._update_existing_usage(existing_usage, usage_params, target_model)
+            if existing_usage:
-            usage = existing_usage
+                if cls._is_usage_finalized(existing_usage):
-        else:
+                    logger.debug(
-            usage = Usage(**usage_params)
+                        "request_id {} 已完成结算，跳过重复记账 (billing_status={})",
-            db.add(usage)
+                        request_id,
-
+                        getattr(existing_usage, "billing_status", None),
-        # 确保 user 和 api_key 在会话中
-        if user and not db.object_session(user):
-            user = db.merge(user)
-        if api_key and not db.object_session(api_key):
-            api_key = db.merge(api_key)
-
-        # 使用原子更新避免并发竞态条件
-        from sqlalchemy import func as sql_func
-        from sqlalchemy import update
-
-        from src.models.database import ApiKey as ApiKeyModel
-        from src.models.database import GlobalModel
-
-        accounted, charge_applied = cls._finalize_usage_billing(
-            db,
-            usage=usage,
-            total_cost=total_cost,
-            status=status,
-        )
-
-        if accounted:
-            # 更新 API 密钥使用量
-            if api_key:
-                values: dict[str, Any] = {
-                    "total_requests": ApiKeyModel.total_requests + 1,
-                    "last_used_at": sql_func.now(),
-                    "updated_at": sql_func.now(),
-                }
-                if charge_applied:
-                    values["total_cost_usd"] = ApiKeyModel.total_cost_usd + Decimal(
-                        str(to_money_decimal(total_cost))
                     )
-                db.execute(update(ApiKeyModel).where(ApiKeyModel.id == api_key.id).values(**values))
+                    return existing_usage
+                logger.debug(
+                    f"request_id {request_id} 已存在，更新现有记录 "
+                    f"(status: {existing_usage.status} -> {status})"
+                )
+                cls._update_existing_usage(existing_usage, usage_params, target_model)
+                usage = existing_usage
+            else:
+                usage = Usage(**usage_params)
+                db.add(usage)
 
-            # 更新 GlobalModel 使用计数
+            # 确保 user 和 api_key 在会话中
-            db.execute(
+            nonlocal user, api_key
-                update(GlobalModel)
+            if user and not db.object_session(user):
-                .where(GlobalModel.name == model)
+                user = db.merge(user)
-                .values(usage_count=GlobalModel.usage_count + 1)
+            if api_key and not db.object_session(api_key):
+                api_key = db.merge(api_key)
+
+            accounted, charge_applied = cls._finalize_usage_billing(
+                db,
+                usage=usage,
+                total_cost=total_cost,
+                status=status,
             )
 
-            # 更新用户-模型调用次数计数器
+            if accounted:
-            cls._increment_user_model_usage(db, user, model)
+                # 更新 API 密钥使用量
+                if api_key:
+                    values: dict[str, Any] = {
+                        "total_requests": ApiKeyModel.total_requests + 1,
+                        "last_used_at": sql_func.now(),
+                        "updated_at": sql_func.now(),
+                    }
+                    if charge_applied:
+                        values["total_cost_usd"] = ApiKeyModel.total_cost_usd + Decimal(
+                            str(to_money_decimal(total_cost))
+                        )
+                    db.execute(
+                        sa_update(ApiKeyModel).where(ApiKeyModel.id == api_key.id).values(**values)
+                    )
 
-            # 更新 Provider 月度使用量（Provider 端真实成本，无论钱包是否扣费）
+                # 更新 GlobalModel 使用计数
-            if provider_id:
-                actual_total_cost = Decimal(str(usage_params["actual_total_cost_usd"]))
                 db.execute(
-                    update(Provider)
+                    sa_update(GlobalModel)
-                    .where(Provider.id == provider_id)
+                    .where(GlobalModel.name == model)
-                    .values(monthly_used_usd=Provider.monthly_used_usd + actual_total_cost)
+                    .values(usage_count=GlobalModel.usage_count + 1)
                 )
 
-            # 更新手动代理节点请求计数（tunnel 节点由心跳上报，不在此处统计）
+                # 更新用户-模型调用次数计数器
-            manual_node_id = _extract_manual_proxy_node_id(metadata)
+                cls._increment_user_model_usage(db, user, model)
-            if manual_node_id:
-                failed = {manual_node_id: 1} if status == "failed" else None
-                _increment_proxy_node_requests(db, {manual_node_id: 1}, failed)
 
-        dispatch_codex_quota_sync_from_response_headers(
+                # 更新 Provider 月度使用量（Provider 端真实成本，无论钱包是否扣费）
-            provider_api_key_id=provider_api_key_id,
+                if provider_id:
-            response_headers=response_headers,
+                    actual_total_cost = Decimal(str(usage_params["actual_total_cost_usd"]))
-            db=db,
+                    db.execute(
-        )
+                        sa_update(Provider)
+                        .where(Provider.id == provider_id)
+                        .values(monthly_used_usd=Provider.monthly_used_usd + actual_total_cost)
+                    )
 
-        # 提交事务
+                # 更新手动代理节点请求计数（tunnel 节点由心跳上报，不在此处统计）
-        try:
+                manual_node_id = _extract_manual_proxy_node_id(metadata)
-            db.commit()
+                if manual_node_id:
-        except Exception as e:
+                    failed = {manual_node_id: 1} if status == "failed" else None
-            logger.error("提交使用记录时出错: {}", e)
+                    _increment_proxy_node_requests(db, {manual_node_id: 1}, failed)
-            db.rollback()
-            raise
 
-        return usage
+            dispatch_codex_quota_sync_from_response_headers(
+                provider_api_key_id=provider_api_key_id,
+                response_headers=response_headers,
+                db=db,
+            )
+
+            # 提交事务
+            try:
+                db.commit()
+            except Exception as e:
+                logger.error("提交使用记录时出错: {}", e)
+                db.rollback()
+                raise
+
+            return usage
+
+        return await asyncio.to_thread(_sync_record)
 
     @classmethod
     async def record_usage_with_custom_cost(

src/services/usage/stream.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import re
 from collections import deque
@@ -518,7 +519,8 @@ class StreamUsageTracker:
                 # yield 后再更新数据库状态（仅第一个 chunk 时执行）
                 if chunk_count == 1 and self.request_id:
                     try:
-                        UsageService.update_usage_status(
+                        await asyncio.to_thread(
+                            UsageService.update_usage_status,
                             db=self.db,
                             request_id=self.request_id,
                             status="streaming",
@@ -1022,7 +1024,8 @@ class EnhancedStreamUsageTracker(StreamUsageTracker):
                 # yield 后再更新数据库状态（仅第一个 chunk 时执行）
                 if chunk_count == 1 and self.request_id:
                     try:
-                        UsageService.update_usage_status(
+                        await asyncio.to_thread(
+                            UsageService.update_usage_status,
                             db=self.db,
                             request_id=self.request_id,
                             status="streaming",