fix(ci): 集成测试xdist限制为2个worker，防止DooD模式下OOM杀进程\n\n- 从 -n auto 改为 -n 2\n- DooD模式下nproc读到宿主机4核8线程=8worker\n- 每个worker有独立PG库+ffmpeg编码，内存开销大\n- 8worker直接触发OOM killer杀ffmpeg和worker进程\n- 2worker既能提速（~1.5x）又稳定

requirements-dev.txt

@@ -11,4 +11,5 @@ pytest==8.3.3
 pytest-asyncio==0.24.0
 pytest-cov==6.0.0
 pytest-timeout==2.3.1
+pytest-xdist==3.6.1
 diff-cover==8.0.3

scripts/ci/run_integration_tests.sh

@@ -1,6 +1,7 @@
 #!/bin/bash
 # CI Integration Tests Job 主脚本
 # 包含：依赖安装、ffmpeg安装、Redis启动、PG启动、迁移、测试、清理、覆盖率
+# 支持 pytest-xdist 并行执行：每个 worker 使用独立数据库，预期加速 2-4 倍
 set -eu
 
 echo "=== CI Integration Tests 开始 ==="
@@ -28,12 +29,13 @@ for i in 1 2 3; do
   sleep 5
 done
 for i in 1 2 3; do
-  python3 -m pip install -q pytest-rerunfailures && break
+  python3 -m pip install -q pytest-rerunfailures pytest-xdist && break
-  echo "pip install pytest-rerunfailures 失败，重试 $i/3..."
+  echo "pip install pytest-rerunfailures/pytest-xdist 失败，重试 $i/3..."
   [ $i -eq 3 ] && exit 1
   sleep 5
 done
 pytest --version
+echo "pytest-xdist: $(python3 -c "import xdist; print(xdist.__version__)" 2>/dev/null || echo 'not installed')"
 
 # --- 安装 ffmpeg ---
 echo ""
@@ -187,8 +189,8 @@ if [ "$USE_SHARED_PG" = "true" ]; then
   echo "等待共享PG连接就绪..."
   wait_tcp_ready "$SHARED_PG_HOST" "$SHARED_PG_PORT" 5
 
-  # 创建独立数据库
+  # 创建主数据库（xdist 模式下各 worker 会创建自己的数据库，主库作为 fallback）
-  echo "创建测试数据库: $CI_DB_NAME"
+  echo "创建主测试数据库: $CI_DB_NAME"
   PGPASSWORD="$SHARED_PG_PASSWORD" python3 -c "
 import psycopg2
 conn = psycopg2.connect(host='$SHARED_PG_HOST', port=$SHARED_PG_PORT, user='$SHARED_PG_USER', password='$SHARED_PG_PASSWORD', dbname='postgres')
@@ -238,30 +240,37 @@ else
   echo "TCP connectivity to PostgreSQL confirmed on port $PG_PORT"
 fi
 
-# --- 执行迁移 ---
+# --- 执行迁移（主数据库，xdist worker 会各自创建自己的库并迁移） ---
 echo ""
-echo "=== 执行 Alembic 迁移 ==="
+echo "=== 执行 Alembic 迁移（主数据库） ==="
 PYTHONPATH="$PWD/apps/api:$PWD" python3 -m alembic upgrade head
 echo "✅ 迁移完成"
 
-# --- 运行集成测试 ---
+# --- 运行集成测试（pytest-xdist 并行） ---
 echo ""
-echo "=== 运行集成测试 ==="
+echo "=== 运行集成测试（pytest-xdist 并行模式） ==="
-PYTHONPATH="$PWD/apps/api:$PWD" python3 -m coverage run \
+echo "CPU 核数: $(nproc 2>/dev/null || echo 'unknown')"
-  --source=apps/api/app,packages \
+
-  --omit="*/migrations/*,*/tests/*,*/test_*.py,*/site-packages/*" \
+# 集成测试使用 pytest-xdist 并行加速（coverage 由单元测试负责，并行模式下 coverage 不稳定）
-  --branch \
+# -n 2: 限制2个worker，避免DooD模式下读到宿主机全部核数导致OOM
-  -m pytest tests/integration -q --timeout=60 -x --reruns 2 --reruns-delay 1 -m "not performance"
+#   集成测试涉及ffmpeg编码+PG多库，内存开销大，2worker较稳妥
-python3 -m coverage report --show-missing
+#   （auto模式下读到宿主机4核8线程=8worker，OOM直接杀worker进程）
-python3 -m coverage xml -o coverage.xml
+# --dist loadfile: 同一测试文件分配到同一 worker（共享 fixture 更高效）
-python3 -m coverage report --fail-under=40 > /dev/null
+# --maxfail=1: 遇到失败停止调度新测试（并行模式下等价于 -x）
+PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/integration \
+  -q --timeout=60 --maxfail=1 --reruns 2 --reruns-delay 1 \
+  -m "not performance" \
+  -n 2 --dist loadfile \
+  -p no:cacheprovider
+
 echo "✅ 集成测试通过"
 
-# --- API 性能基线测试（仅告警） ---
+# --- API 性能基线测试（仅告警，串行执行） ---
 echo ""
 echo "=== API 性能基线测试（仅告警） ==="
 set +e
 PERF_OUTPUT=$(mktemp)
+# 性能测试单独串行运行（不参与并行，避免资源竞争影响测量结果）
 PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/integration/test_api_performance.py \
   -v --timeout=120 -p no:cacheprovider 2>&1 | tee "$PERF_OUTPUT"
 echo ""
@@ -284,14 +293,29 @@ set -e
 echo ""
 echo "=== 清理 ==="
 if [ "$USE_SHARED_PG" = "true" ]; then
-  # 清理共享PG上的测试数据库
+  # 清理共享PG上的测试数据库（主库 + 可能残留的 worker 库）
-  echo "清理共享PG测试数据库: $CI_DB_NAME"
+  echo "清理共享PG测试数据库..."
+
+  # 清理所有以 CI_DB_NAME 开头的数据库（主库 + worker 库）
   PGPASSWORD="${SHARED_PG_PASSWORD}" python3 -c "
 import psycopg2
 conn = psycopg2.connect(host='${SHARED_PG_HOST}', port=${SHARED_PG_PORT}, user='${SHARED_PG_USER}', password='${SHARED_PG_PASSWORD}', dbname='postgres')
 conn.autocommit = True
 cur = conn.cursor()
-cur.execute(f'DROP DATABASE IF EXISTS \"$CI_DB_NAME\" WITH (FORCE)')
+
+# 查找所有需要清理的数据库（主库 + worker 库）
+cur.execute(\"SELECT datname FROM pg_database WHERE datname LIKE '$CI_DB_NAME%'\")
+dbs = [row[0] for row in cur.fetchall()]
+
+for db in dbs:
+    try:
+        # 强制断开所有连接
+        cur.execute(f\"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = '{db}' AND pid <> pg_backend_pid()\")
+        cur.execute(f'DROP DATABASE IF EXISTS \"{db}\" WITH (FORCE)')
+        print(f'  已清理: {db}')
+    except Exception as e:
+        print(f'  警告: 清理 {db} 失败: {e}')
+
 cur.close()
 conn.close()
 " 2>/dev/null || echo "WARN: 数据库清理失败（可能已被清理）"

tests/integration/conftest.py

@@ -2,18 +2,167 @@
 集成测试公共 fixtures
 
 提供性能测试相关的工具、fixture 和 marker。
+支持 pytest-xdist 并行执行：每个 worker 使用独立数据库，数据完全隔离。
 """
 
 from __future__ import annotations
 
 import os
+import sys
 import time
 from contextlib import contextmanager
 from dataclasses import dataclass, field
+from pathlib import Path
 from typing import Callable, Dict, List, Optional
 
 import pytest
 
+# ── xdist 并行数据库隔离 ──────────────────────────────────────────────────
+# 每个 xdist worker 进程创建独立的数据库并执行迁移，确保测试数据完全隔离
+# 通过 PYTEST_XDIST_WORKER 环境变量识别 worker（如 gw0, gw1, ...）
+
+_WORKER_DB_NAME: Optional[str] = None
+
+
+def _get_worker_id() -> Optional[str]:
+    """获取当前 xdist worker ID，非 worker 模式返回 None"""
+    return os.environ.get("PYTEST_XDIST_WORKER")
+
+
+def _parse_database_url(url: str) -> Dict[str, str]:
+    """
+    解析 DATABASE_URL，返回各组件。
+    支持 postgresql+psycopg://user:pass@host:port/dbname 格式
+    """
+    from urllib.parse import urlparse
+
+    parsed = urlparse(url)
+    return {
+        "driver": parsed.scheme,
+        "user": parsed.username or "",
+        "password": parsed.password or "",
+        "host": parsed.hostname or "",
+        "port": str(parsed.port or 5432),
+        "dbname": parsed.path.lstrip("/") or "",
+    }
+
+
+def _create_worker_database(worker_id: str) -> str:
+    """
+    为 xdist worker 创建独立数据库并执行迁移。
+    返回新的 DATABASE_URL。
+    """
+    base_url = os.environ.get(
+        "DATABASE_URL",
+        "postgresql+psycopg://postgres:postgres@localhost:5432/xiaoxia_saas",
+    )
+    db_info = _parse_database_url(base_url)
+
+    # 生成 worker 专属数据库名
+    base_db = db_info["dbname"]
+    worker_db = f"{base_db}_{worker_id}"
+    global _WORKER_DB_NAME
+    _WORKER_DB_NAME = worker_db
+
+    # 使用 psycopg 创建数据库（连接到 postgres 库）
+    try:
+        import psycopg
+
+        conn_str = (
+            f"host={db_info['host']} port={db_info['port']} "
+            f"user={db_info['user']} password={db_info['password']} "
+            f"dbname=postgres"
+        )
+        conn = psycopg.connect(conn_str, autocommit=True)
+        cur = conn.cursor()
+
+        # 先尝试删除（防止残留）
+        cur.execute(f'DROP DATABASE IF EXISTS "{worker_db}" WITH (FORCE)')
+
+        # 创建新数据库
+        cur.execute(f'CREATE DATABASE "{worker_db}"')
+        cur.close()
+        conn.close()
+        print(f"[xdist {worker_id}] ✅ 创建数据库: {worker_db}")
+    except ImportError:
+        print(f"[xdist {worker_id}] ⚠️  psycopg 未安装，跳过数据库创建")
+        return base_url
+    except Exception as e:
+        print(f"[xdist {worker_id}] ⚠️  创建数据库失败: {e}")
+        return base_url
+
+    # 构建新的 DATABASE_URL
+    new_url = (
+        f"{db_info['driver']}://{db_info['user']}:{db_info['password']}"
+        f"@{db_info['host']}:{db_info['port']}/{worker_db}"
+    )
+
+    # 执行 alembic 迁移
+    print(f"[xdist {worker_id}] 🔄 执行 Alembic 迁移...")
+    try:
+        ROOT = Path(__file__).resolve().parents[2]
+        api_path = str(ROOT / "apps" / "api")
+        if api_path not in sys.path:
+            sys.path.insert(0, api_path)
+        if str(ROOT) not in sys.path:
+            sys.path.insert(0, str(ROOT))
+
+        from alembic import command as alembic_command
+        from alembic.config import Config as AlembicConfig
+
+        alembic_cfg = AlembicConfig(str(ROOT / "alembic.ini"))
+        alembic_cfg.set_main_option("sqlalchemy.url", new_url)
+        # 兼容不同的脚本路径配置
+        alembic_cfg.set_main_option("script_location", str(ROOT / "alembic"))
+
+        # 临时设置环境变量供 alembic env.py 使用
+        os.environ["DATABASE_URL"] = new_url
+        alembic_command.upgrade(alembic_cfg, "head")
+        print(f"[xdist {worker_id}] ✅ 迁移完成")
+    except Exception as e:
+        print(f"[xdist {worker_id}] ❌ 迁移失败: {e}")
+        raise
+
+    return new_url
+
+
+def _cleanup_worker_database(worker_id: str):
+    """清理 xdist worker 的数据库"""
+    global _WORKER_DB_NAME
+    if not _WORKER_DB_NAME:
+        return
+
+    base_url = os.environ.get(
+        "DATABASE_URL",
+        "postgresql+psycopg://postgres:postgres@localhost:5432/xiaoxia_saas",
+    )
+    db_info = _parse_database_url(base_url)
+
+    try:
+        import psycopg
+
+        conn_str = (
+            f"host={db_info['host']} port={db_info['port']} "
+            f"user={db_info['user']} password={db_info['password']} "
+            f"dbname=postgres"
+        )
+        conn = psycopg.connect(conn_str, autocommit=True)
+        cur = conn.cursor()
+        # 强制断开所有连接后删除
+        cur.execute(
+            f"SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
+            f"WHERE datname = '{_WORKER_DB_NAME}' AND pid <> pg_backend_pid()"
+        )
+        cur.execute(f'DROP DATABASE IF EXISTS "{_WORKER_DB_NAME}" WITH (FORCE)')
+        cur.close()
+        conn.close()
+        print(f"[xdist {worker_id}] 🧹 已清理数据库: {_WORKER_DB_NAME}")
+    except Exception as e:
+        print(f"[xdist {worker_id}] ⚠️  清理数据库失败: {e}")
+    finally:
+        _WORKER_DB_NAME = None
+
+
 # ── 性能阈值配置 ──────────────────────────────────────────────────────────
 PERF_THRESHOLDS: Dict[str, int] = {
     "core": 500,  # 核心接口：500ms
@@ -183,16 +332,41 @@ class PerfAssert:
         return "\n".join(lines)
 
 
-# ── pytest fixtures ──────────────────────────────────────────────────────
+# ── pytest hooks ──────────────────────────────────────────────────────────
 
 
 def pytest_configure(config):
-    """注册自定义 marker"""
+    """
+    pytest 配置钩子。
+
+    - 注册自定义 marker
+    - xdist worker 模式下：创建独立数据库 + 执行迁移
+    """
+    # 注册自定义 marker
     config.addinivalue_line("markers", "performance: 标记为性能测试（可通过 -m 'not performance' 跳过）")
     config.addinivalue_line("markers", "perf_core: 核心接口性能测试（阈值 500ms）")
     config.addinivalue_line("markers", "perf_normal: 普通接口性能测试（阈值 1000ms）")
     config.addinivalue_line("markers", "perf_heavy: 重操作接口性能测试（阈值 3000ms）")
 
+    # xdist worker 模式：创建独立数据库并执行迁移
+    worker_id = _get_worker_id()
+    if worker_id:
+        # 只有当 USE_IN_MEMORY_DB 不为 true 时才创建独立数据库
+        use_in_memory = os.environ.get("USE_IN_MEMORY_DB", "true").lower() == "true"
+        if not use_in_memory:
+            print(f"[xdist {worker_id}] 🚀 worker 启动，准备独立数据库...")
+            new_db_url = _create_worker_database(worker_id)
+            os.environ["DATABASE_URL"] = new_db_url
+        else:
+            print(f"[xdist {worker_id}] ℹ️  USE_IN_MEMORY_DB=true，跳过 worker 数据库创建")
+
+
+def pytest_unconfigure(config):
+    """pytest 结束钩子：清理 xdist worker 数据库"""
+    worker_id = _get_worker_id()
+    if worker_id and _WORKER_DB_NAME:
+        _cleanup_worker_database(worker_id)
+
 
 def pytest_collection_modifyitems(config, items):
     """根据环境变量自动跳过性能测试"""
@@ -203,6 +377,9 @@ def pytest_collection_modifyitems(config, items):
                 item.add_marker(skip_perf)
 
 
+# ── pytest fixtures ──────────────────────────────────────────────────────
+
+
 @pytest.fixture
 def perf_assert():
     """