fix(P1-E): 启动时标 failed 残留 pending/running ingest_jobs

lifespan 加 _fail_stale_ingest_jobs():UPDATE ingest_jobs SET status=failed
WHERE status IN ('pending','running');异常仅记 warning 不阻断启动。
顺手修复 0021(op.text→字符串)/0023(索引名 leason→league)/0024(type_unique→type_)
三个待执行迁移 bug。

测试 test_p1_e_stale_ingest_jobs(2/2 mock);全量 315 通过。

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',
     )
 
 

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')
+    # 原始索引名拼写为 leason(历史遗留),按实际库名删除
-    op.create_index('ix_standings_league_season_pos', 'standings', ['league_id', 'season', 'position'])
+    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'],

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(

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 不共享。

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