diff --git a/alembic/versions/0021_match_source_event_id_unique.py b/alembic/versions/0021_match_source_event_id_unique.py index c4013da..611130b 100644 --- a/alembic/versions/0021_match_source_event_id_unique.py +++ b/alembic/versions/0021_match_source_event_id_unique.py @@ -30,7 +30,7 @@ def upgrade() -> None: 'matches', ['source_event_id'], unique=True, - postgresql_where=op.text('source_event_id IS NOT NULL'), + postgresql_where='source_event_id IS NOT NULL', ) diff --git a/alembic/versions/0023_standings_append_only.py b/alembic/versions/0023_standings_append_only.py index 7e67149..15cf94e 100644 --- a/alembic/versions/0023_standings_append_only.py +++ b/alembic/versions/0023_standings_append_only.py @@ -34,8 +34,9 @@ def upgrade() -> None: # 2) 去旧唯一约束,加新唯一约束(league, season, team, available_at) op.drop_constraint('uq_standings_league_season_team', 'standings', type_='unique') - op.drop_index('ix_standings_leason_season_pos', table_name='standings') - op.create_index('ix_standings_league_season_pos', 'standings', ['league_id', 'season', 'position']) + # 原始索引名拼写为 leason(历史遗留),按实际库名删除 + op.drop_index('ix_standings_league_season_pos', table_name='standings', if_exists=True) + op.create_index('ix_standings_league_season_pos_v2', 'standings', ['league_id', 'season', 'position']) op.create_unique_constraint( 'uq_standings_league_season_team_available', 'standings', ['league_id', 'season', 'team_id', 'available_at'], diff --git a/alembic/versions/0024_prediction_idempotent_fingerprint.py b/alembic/versions/0024_prediction_idempotent_fingerprint.py index 7648911..120fefd 100644 --- a/alembic/versions/0024_prediction_idempotent_fingerprint.py +++ b/alembic/versions/0024_prediction_idempotent_fingerprint.py @@ -23,7 +23,7 @@ def upgrade() -> None: # 移除旧唯一约束(match, provider, model, mode, run_type) op.drop_constraint( 'uq_predictions_match_provider_model_mode_run_type', - 'predictions', type_unique=True, + 'predictions', type_='unique', ) # P0-03: partial unique on input_hash(非空时唯一) op.create_index( diff --git a/src/api/app.py b/src/api/app.py index 4525476..ab359f2 100644 --- a/src/api/app.py +++ b/src/api/app.py @@ -14,6 +14,35 @@ from src.core.config import settings logger = logging.getLogger(__name__) +async def _fail_stale_ingest_jobs() -> None: + """P1-E: 启动时将上次遗留的 pending/running ingest_jobs 标 failed。 + + 进程异常退出(重启/OOM)会导致 ingest_jobs 残留为 pending/running, + 这些任务实际已不在执行,启动时一次性标 failed 避免永久"执行中"。 + 尽力而为:失败只记 warning,不阻断启动。 + """ + from datetime import datetime, timezone + + from sqlalchemy import update + + from src.db.base import AsyncSessionLocal + from src.db.models import IngestJob + + try: + async with AsyncSessionLocal() as session: + stmt = ( + update(IngestJob) + .where(IngestJob.status.in_(["pending", "running"])) + .values(status="failed", error="进程重启:任务被终止", finished_at=datetime.now(timezone.utc)) + ) + result = await session.execute(stmt) + await session.commit() + if result.rowcount: + logger.info("P1-E: 已将 %d 条残留 pending/running ingest_jobs 标 failed", result.rowcount) + except Exception: + logger.warning("P1-E: 清理残留 ingest_jobs 失败,不影响启动", exc_info=True) + + @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: from src.db.base import init_db @@ -30,6 +59,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: await migrate_plaintext_sensitive_settings() # 明文敏感配置 → 加密(幂等) await ensure_admin_password_hashed() # .env 明文密码 → scrypt 哈希(幂等) await assert_security_on_startup() # 启动安全校验(生产拒绝/开发警告) + await _fail_stale_ingest_jobs() # P1-E: 上次遗留的 pending/running 标 failed # D7(工程债): 进程内限流(_RateLimiter)与 KeyRing 均为单进程状态; # 多 worker 部署时各进程独立计数,限流阈值会按 worker 数放大、KeyRing 不共享。 diff --git a/tests/test_p1_e_stale_ingest_jobs.py b/tests/test_p1_e_stale_ingest_jobs.py new file mode 100644 index 0000000..82380c3 --- /dev/null +++ b/tests/test_p1_e_stale_ingest_jobs.py @@ -0,0 +1,80 @@ +"""P1-E 回归测试: 启动时将过期 pending/running ingest_jobs 标 failed。 + +运行: pytest tests/test_p1_e_stale_ingest_jobs.py -v +(使用 mock session,验证 UPDATE 过滤条件与 SET 值。) +""" +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +import src.api.app as app_mod + + +def _make_fake_session(commit_out: dict): + session = MagicMock() + session.__aenter__ = MagicMock(return_value=session) + session.__aexit__ = MagicMock(return_value=False) + + async def execute(stmt): + commit_out["sql"] = str(stmt) + result = MagicMock() + result.rowcount = 2 + return result + + async def commit(): + commit_out["commit"] = commit_out.get("commit", 0) + 1 + + session.execute = execute + session.commit = commit + return session + + +@pytest.mark.asyncio +async def test_fail_stale_updates_pending_and_running_only(): + """P1-E: UPDATE 必须过滤 status IN ('pending','running'),SET status=failed。""" + out = {} + fake = _make_fake_session(out) + + class FakeSessionLocal: + def __init__(self): + self._s = fake + + async def __aenter__(self): + return self._s + + async def __aexit__(self, *a): + return None + + # patch src.db.base.AsyncSessionLocal(函数内 from-import 每次调用从此取) + import src.db.base as _base + + with patch.object(_base, "AsyncSessionLocal", FakeSessionLocal): + await app_mod._fail_stale_ingest_jobs() + + sql = out.get("sql", "").lower() + # WHERE 子句过滤 status IN (绑定参数,SQLAlchemy 用 __[postcompile_x] 占位) + assert "status in" in sql, f"SQL 缺少 status IN 过滤: {sql}" + # SET status=failed + assert "status=:status" in sql or "status=" in sql, f"SQL 缺少 status 更新: {sql}" + # 应清理成功(commit 被调用) + assert out.get("commit", 0) >= 1, f"应已提交,实际 commit 调用: {out}" + + +@pytest.mark.asyncio +async def test_fail_stale_does_not_raise_on_db_error(): + """P1-E: DB 异常不得阻断启动(仅记 warning)。""" + class BoomSessionLocal: + async def __aenter__(self): + raise RuntimeError("DB down") + + async def __aexit__(self, *a): + return None + + import src.db.base as _base + + with patch.object(_base, "AsyncSessionLocal", BoomSessionLocal): + # 不应抛异常 + await app_mod._fail_stale_ingest_jobs() + assert True