diff --git a/app/api/controllers/risk.py b/app/api/controllers/risk.py index 0cc2da1..878debe 100644 --- a/app/api/controllers/risk.py +++ b/app/api/controllers/risk.py @@ -1,11 +1,11 @@ """风控只读查询接口。""" import json -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Awaitable, Callable from datetime import datetime, time from typing import Any -from fastapi import APIRouter, Depends, File, Path, Request, UploadFile +from fastapi import APIRouter, Depends, File, Header, Path, Request, UploadFile from sqlalchemy.ext.asyncio import AsyncSession from starlette.responses import StreamingResponse @@ -27,6 +27,7 @@ from app.api.schemas.risk import ( from app.core.contracts import RequestContext from app.core.errors import SseNotAcceptableError from app.infrastructure.db import mysql_scan_lock +from app.service.api_transaction_service import ApiTransactionService from app.service.risk_action_service import RiskActionService from app.service.risk_daily_report_mail_service import RiskDailyReportMailService from app.service.risk_daily_report_service import RiskDailyReportService @@ -42,6 +43,26 @@ router = APIRouter( ) +async def _idempotent_write( + session: AsyncSession, + context: RequestContext, + key: str | None, + scope: str, + body: Any, + action: Callable[[AsyncSession], Awaitable[dict[str, Any]]], +) -> dict[str, Any]: + """风控写接口的统一幂等入口（`docs/05` §5.1、§5.2、§13.2）。 + + `scope` 用**实际**路径（含 `alert_no`）：§5.1 的幂等范围是 + `user_id + method + normalized_path + idempotency_key`，把路径参数折成模板会让 + 同一个键在不同预警之间互相回放 —— 那是把两次不同资源的操作当成一次。 + + 幂等记录与业务写入同事务（见 `ApiTransactionService.execute_in`），重复请求直接 + 回放 `response_json`，不会二次驱动状态机。 + """ + return await ApiTransactionService().execute_in(session, context, scope, key, body, action) + + @router.get("/overview") async def risk_overview( context: RequestContext = Depends(build_request_context), # noqa: B008 @@ -65,15 +86,21 @@ async def list_risk_alerts( async def scan_risk_alerts( context: RequestContext = Depends(build_request_context), # noqa: B008 session: AsyncSession = Depends(get_session), # noqa: B008 + key: str | None = Header(default=None, alias="Idempotency-Key"), ) -> dict[str, object]: - # 手工触发的扫描必须与定时扫描互斥，否则两条路径会同时查不到重复、同时插入。 - # 锁加在**入口层**而不是 `RiskScanService.scan()` 内部：`GET_LOCK` 是连接级的， - # 而调度器已在它自己的 session 上持锁 —— 被两个入口共用的服务方法若再取同一把锁， - # 取锁的连接不是持锁的那一个、必然失败，会**把定时扫描自己挡死**。 - async with mysql_scan_lock() as acquired: - if not acquired: - raise RiskScanBusyError("规则扫描正在执行，请稍后重试") - data = await RiskScanService(session).scan(context) + async def run(inner: AsyncSession) -> dict[str, Any]: + # 手工触发的扫描必须与定时扫描互斥，否则两条路径会同时查不到重复、同时插入。 + # 锁加在**入口层**而不是 `RiskScanService.scan()` 内部：`GET_LOCK` 是连接级的， + # 而调度器已在它自己的 session 上持锁 —— 被两个入口共用的服务方法若再取同一把锁， + # 取锁的连接不是持锁的那一个、必然失败，会**把定时扫描自己挡死**。 + async with mysql_scan_lock() as acquired: + if not acquired: + raise RiskScanBusyError("规则扫描正在执行，请稍后重试") + return await RiskScanService(inner).scan(context) + + data = await _idempotent_write( + session, context, key, "POST /api/v1/risk/alerts/scan", {}, run + ) return _envelope(data, context) @@ -82,8 +109,16 @@ async def acknowledge_risk_alert( alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"), context: RequestContext = Depends(build_request_context), # noqa: B008 session: AsyncSession = Depends(get_session), # noqa: B008 + key: str | None = Header(default=None, alias="Idempotency-Key"), ) -> dict[str, object]: - data = await RiskActionService(session).acknowledge(alert_no, context) + data = await _idempotent_write( + session, + context, + key, + f"POST /api/v1/risk/alerts/{alert_no}/acknowledgements", + {}, + lambda inner: RiskActionService(inner).acknowledge(alert_no, context), + ) return _envelope(data, context) @@ -92,8 +127,16 @@ async def investigate_risk_alert( alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"), context: RequestContext = Depends(build_request_context), # noqa: B008 session: AsyncSession = Depends(get_session), # noqa: B008 + key: str | None = Header(default=None, alias="Idempotency-Key"), ) -> dict[str, object]: - data = await RiskActionService(session).investigate(alert_no, context) + data = await _idempotent_write( + session, + context, + key, + f"POST /api/v1/risk/alerts/{alert_no}/investigations", + {}, + lambda inner: RiskActionService(inner).investigate(alert_no, context), + ) return _envelope(data, context) @@ -103,8 +146,16 @@ async def exclude_risk_alert( alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"), context: RequestContext = Depends(build_request_context), # noqa: B008 session: AsyncSession = Depends(get_session), # noqa: B008 + key: str | None = Header(default=None, alias="Idempotency-Key"), ) -> dict[str, object]: - data = await RiskActionService(session).exclude(alert_no, payload.reason, context) + data = await _idempotent_write( + session, + context, + key, + f"POST /api/v1/risk/alerts/{alert_no}/exclusions", + {"reason": payload.reason}, + lambda inner: RiskActionService(inner).exclude(alert_no, payload.reason, context), + ) return _envelope(data, context) @@ -114,8 +165,16 @@ async def resolve_risk_alert( alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"), context: RequestContext = Depends(build_request_context), # noqa: B008 session: AsyncSession = Depends(get_session), # noqa: B008 + key: str | None = Header(default=None, alias="Idempotency-Key"), ) -> dict[str, object]: - data = await RiskActionService(session).resolve(alert_no, payload.resolution, context) + data = await _idempotent_write( + session, + context, + key, + f"POST /api/v1/risk/alerts/{alert_no}/resolutions", + {"resolution": payload.resolution}, + lambda inner: RiskActionService(inner).resolve(alert_no, payload.resolution, context), + ) return _envelope(data, context) @@ -125,8 +184,16 @@ async def escalate_risk_alert( alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"), context: RequestContext = Depends(build_request_context), # noqa: B008 session: AsyncSession = Depends(get_session), # noqa: B008 + key: str | None = Header(default=None, alias="Idempotency-Key"), ) -> dict[str, object]: - data = await RiskActionService(session).escalate(alert_no, payload.reason, context) + data = await _idempotent_write( + session, + context, + key, + f"POST /api/v1/risk/alerts/{alert_no}/escalations", + {"reason": payload.reason}, + lambda inner: RiskActionService(inner).escalate(alert_no, payload.reason, context), + ) return _envelope(data, context) diff --git a/app/service/api_transaction_service.py b/app/service/api_transaction_service.py index 2ce538a..4cdc3ae 100644 --- a/app/service/api_transaction_service.py +++ b/app/service/api_transaction_service.py @@ -41,3 +41,49 @@ class ApiTransactionService: await session.execute(table.update().where(table.c.id == row["id"]) .values(response_json=response)) return response + + async def execute_in( + self, + session: AsyncSession, + context: RequestContext, + scope: str, + key: str | None, + body: Any, + action: Callable[[AsyncSession], Awaitable[dict[str, Any]]], + ) -> dict[str, Any]: + """`execute` 的"用调用方事务"版本，供 B 类业务写接口复用。 + + `execute` 自己开 `SessionFactory()` 和 `session.begin()`；但业务 Action Service + 普遍在收尾时自己 `commit()`（例如 `RiskActionService._finish`），把它套进外层 + `session.begin()` 就成了"内层提交外层事务"。这里改为在**调用方传入的 session** + 上读写幂等记录，幂等记录因此与业务写入同处一个事务（`docs/05` §5.2）。 + + 并发同键由 `SELECT … FOR UPDATE` 串行化：后到的请求要么直接回放 + `response_json`，要么在拿到锁后看到同一请求哈希而继续执行。 + """ + if key is None or not 16 <= len(key) <= 128 or not key.isascii(): + raise ValidationAgentError("必须提供 16-128 位 ASCII Idempotency-Key") + table = await PlatformRepository(session).table("api_request_receipt") + statement = insert(table).values( + user_id=int(context.user_id), + scope_hash=digest(scope), + idempotency_key=key, + request_hash=digest(body), + ) + await session.execute(statement.on_duplicate_key_update(id=table.c.id)) + row = (await session.execute(select(table).where( + table.c.user_id == int(context.user_id), + table.c.scope_hash == digest(scope), + table.c.idempotency_key == key).with_for_update())).mappings().one() + if row["request_hash"] != digest(body): + raise IdempotencyConflictError("同一幂等键对应不同请求") + if row["response_json"] is not None: + return dict(row["response_json"]) + response = await action(session) + await session.execute(table.update().where(table.c.id == row["id"]) + .values(response_json=response)) + # 业务 Action 多半已经提交了自己的写入（风控就是这样），所以这里必须再提交 + # 一次才能把刚写回的 `response_json` 落库；若业务没有提交，则这一步同时提交 + # 业务写入与幂等记录。 + await session.commit() + return response diff --git a/tests/integration/test_risk_idempotency_mysql.py b/tests/integration/test_risk_idempotency_mysql.py new file mode 100644 index 0000000..e259dc4 --- /dev/null +++ b/tests/integration/test_risk_idempotency_mysql.py @@ -0,0 +1,178 @@ +"""风控写接口的幂等语义（`docs/05` §5.1、§5.2；`docs/25` P3 #22）。 + +两级覆盖： + +1. `ApiTransactionService.execute_in` 本身 —— 同键同正文回放原响应且**不重复执行**、 + 同键不同正文返回 `409 IDEMPOTENCY_CONFLICT`、缺键直接拒绝； +2. Controller 接线 —— 重复 POST 同一个风控处置端点时，Action Service 只被调用一次。 + +刻意不去驱动真实状态机：处置动作会改预警状态，测试不该污染演示数据。这里用计数替身 +验证"第二次请求没有落到业务逻辑上"，这才是幂等要保证的事情。 +""" + +from collections.abc import AsyncIterator +from typing import Any +from uuid import uuid4 + +import httpx +import pytest +from sqlalchemy import delete +from sqlalchemy.ext.asyncio import AsyncSession + +from app.api.controllers import risk as risk_controller +from app.api.dependencies.auth import build_request_context +from app.api.dependencies.database import get_session +from app.core.contracts import RequestContext +from app.core.errors import IdempotencyConflictError, ValidationAgentError +from app.infrastructure.db import SessionFactory +from app.main import app +from app.repository.platform_repository import PlatformRepository +from app.service.api_transaction_service import ApiTransactionService + +SCOPE = "POST /api/v1/risk/alerts/ALERT-IDEM-TEST/acknowledgements" +EXCLUSION_SCOPE = "POST /api/v1/risk/alerts/ALERT-IDEM-TEST/exclusions" +ACK_PATH = "/api/v1/risk/alerts/ALERT-IDEM-TEST/acknowledgements" +EXCLUSION_PATH = "/api/v1/risk/alerts/ALERT-IDEM-TEST/exclusions" + + +async def override_context() -> RequestContext: + return RequestContext( + user_id="990000002", + trace_id=str(uuid4()), + roles=("risk_operator",), + permissions=("risk:alert:read", "risk:alert:write"), + data_scope="all", + ) + + +async def override_session() -> AsyncIterator[AsyncSession]: + async with SessionFactory() as session: + yield session + + +async def purge(*keys: str) -> None: + async with SessionFactory() as session, session.begin(): + table = await PlatformRepository(session).table("api_request_receipt") + for key in keys: + await session.execute(delete(table).where(table.c.idempotency_key == key)) + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_execute_in_replays_without_running_action_twice() -> None: + key = f"risk-idem-{uuid4()}" + context = await override_context() + calls: list[int] = [] + + async def action(_session: AsyncSession) -> dict[str, Any]: + calls.append(1) + return {"alert_no": "ALERT-IDEM-TEST", "status": "待处理"} + + try: + async with SessionFactory() as session: + service = ApiTransactionService() + first = await service.execute_in(session, context, SCOPE, key, {}, action) + second = await service.execute_in(session, context, SCOPE, key, {}, action) + + assert first == second == {"alert_no": "ALERT-IDEM-TEST", "status": "待处理"} + assert len(calls) == 1, "重复请求必须回放 response_json，而不是再次执行 action" + finally: + await purge(key) + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_execute_in_conflicts_on_same_key_with_different_body() -> None: + key = f"risk-idem-{uuid4()}" + context = await override_context() + + async def action(_session: AsyncSession) -> dict[str, Any]: + return {"ok": True} + + try: + async with SessionFactory() as session: + service = ApiTransactionService() + await service.execute_in(session, context, SCOPE, key, {"reason": "第一次"}, action) + with pytest.raises(IdempotencyConflictError): + await service.execute_in( + session, context, SCOPE, key, {"reason": "第二次"}, action + ) + finally: + await purge(key) + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_execute_in_rejects_missing_or_short_key() -> None: + context = await override_context() + + async def action(_session: AsyncSession) -> dict[str, Any]: + raise AssertionError("缺键时不应执行 action") + + async with SessionFactory() as session: + service = ApiTransactionService() + for bad in (None, "too-short", "带中文字符的-key-1234567890"): + with pytest.raises(ValidationAgentError): + await service.execute_in(session, context, SCOPE, bad, {}, action) + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_repeated_risk_post_calls_action_service_once(monkeypatch) -> None: + """端到端：同一 `Idempotency-Key` 重复 POST 只触发一次处置逻辑。 + + 用 `httpx.ASGITransport` 而不是 `TestClient`：后者自建事件循环，测试结束后的 + `SessionFactory` 清理会落在另一个循环上，连接池析构时报 + `AttributeError: 'NoneType' object has no attribute 'send'`。 + """ + calls: list[str] = [] + + class CountingActionService: + def __init__(self, _session: Any) -> None: + pass + + async def acknowledge(self, alert_no: str, _context: RequestContext) -> dict[str, Any]: + calls.append(alert_no) + return {"alert_no": alert_no, "status": "待处理", "ack_status": "已确认"} + + async def exclude( + self, alert_no: str, reason: str, _context: RequestContext + ) -> dict[str, Any]: + calls.append(f"exclude:{reason}") + return {"alert_no": alert_no, "status": "已排除", "handle_result": reason} + + monkeypatch.setattr(risk_controller, "RiskActionService", CountingActionService) + ack_key = f"risk-ack-{uuid4()}" + exclude_key = f"risk-exclude-{uuid4()}" + app.dependency_overrides[build_request_context] = override_context + app.dependency_overrides[get_session] = override_session + try: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient( + transport=transport, base_url="http://testserver" + ) as client: + first = await client.post(ACK_PATH, headers={"Idempotency-Key": ack_key}) + replay = await client.post(ACK_PATH, headers={"Idempotency-Key": ack_key}) + excluded = await client.post( + EXCLUSION_PATH, + json={"reason": "客户本人确认"}, + headers={"Idempotency-Key": exclude_key}, + ) + conflict = await client.post( + EXCLUSION_PATH, + json={"reason": "换了理由"}, + headers={"Idempotency-Key": exclude_key}, + ) + + assert first.status_code == 200 + assert replay.status_code == 200 + # 只比 `data`：`meta.trace_id` 标识的是**本次**请求，重放也必须换一个新的。 + assert replay.json()["data"] == first.json()["data"] + assert excluded.status_code == 200 + assert conflict.status_code == 409 + assert conflict.json()["error"]["code"] == "IDEMPOTENCY_CONFLICT" + assert calls == ["ALERT-IDEM-TEST", "exclude:客户本人确认"] + finally: + app.dependency_overrides.pop(build_request_context, None) + app.dependency_overrides.pop(get_session, None) + await purge(ack_key, exclude_key) diff --git a/tests/unit/api/test_risk_controller.py b/tests/unit/api/test_risk_controller.py index 69d9ea0..995789e 100644 --- a/tests/unit/api/test_risk_controller.py +++ b/tests/unit/api/test_risk_controller.py @@ -8,6 +8,8 @@ from app.api.dependencies.database import get_session from app.core.contracts import RequestContext from app.main import create_app +IDEMPOTENCY = {"Idempotency-Key": "risk-controller-test-0001"} + class StubRiskQueryService: def __init__(self, _session: Any) -> None: @@ -31,6 +33,25 @@ class StubRiskQueryService: return {"items": [{"source": source}], "next_cursor": None, "has_more": False} +class StubApiTransactionService: + """幂等透传替身：直接执行 action，不写 `api_request_receipt`。 + + 单测不连库，幂等读写的真实语义由 `tests/integration` 覆盖；这里只保证请求体与 + 状态码断言不被数据库依赖污染。 + """ + + async def execute_in( + self, + session: Any, + _context: RequestContext, + _scope: str, + _key: str | None, + _body: Any, + action: Any, + ) -> dict[str, Any]: + return await action(session) + + class StubRiskScanService: def __init__(self, _session: Any) -> None: pass @@ -120,7 +141,7 @@ class StubRiskDailyReportMailService: return {"status": "dry_run", "recipient_count": len(recipients)} -def authenticated_client(monkeypatch) -> TestClient: +def authenticated_client(monkeypatch, *, idempotent: bool = True) -> TestClient: async def context() -> RequestContext: return RequestContext( user_id="990000002", @@ -145,6 +166,12 @@ def authenticated_client(monkeypatch) -> TestClient: "RiskNotificationService", StubRiskNotificationService, ) + if idempotent: + monkeypatch.setattr( + risk_controller, + "ApiTransactionService", + StubApiTransactionService, + ) monkeypatch.setattr( risk_controller, "RiskDailyReportService", @@ -210,7 +237,7 @@ def test_evidence_route_and_page_limit_are_enforced(monkeypatch) -> None: def test_scan_route_returns_success_envelope(monkeypatch) -> None: with authenticated_client(monkeypatch) as client: - response = client.post("/api/v1/risk/alerts/scan") + response = client.post("/api/v1/risk/alerts/scan", headers=IDEMPOTENCY) assert response.status_code == 200 assert response.json() == { @@ -232,23 +259,31 @@ def test_action_routes_require_authentication() -> None: def test_action_routes_use_success_envelope_and_validate_body(monkeypatch) -> None: with authenticated_client(monkeypatch) as client: - acknowledged = client.post("/api/v1/risk/alerts/ALERT-001/acknowledgements") - investigated = client.post("/api/v1/risk/alerts/ALERT-001/investigations") + acknowledged = client.post( + "/api/v1/risk/alerts/ALERT-001/acknowledgements", headers=IDEMPOTENCY + ) + investigated = client.post( + "/api/v1/risk/alerts/ALERT-001/investigations", headers=IDEMPOTENCY + ) excluded = client.post( "/api/v1/risk/alerts/ALERT-001/exclusions", json={"reason": "客户本人确认"}, + headers=IDEMPOTENCY, ) resolved = client.post( "/api/v1/risk/alerts/ALERT-001/resolutions", json={"resolution": "已核实并留痕"}, + headers=IDEMPOTENCY, ) escalated = client.post( "/api/v1/risk/alerts/ALERT-001/escalations", json={"reason": "需要高级复核"}, + headers=IDEMPOTENCY, ) invalid = client.post( "/api/v1/risk/alerts/ALERT-001/exclusions", json={"reason": " "}, + headers=IDEMPOTENCY, ) assert acknowledged.status_code == 200 @@ -259,6 +294,21 @@ def test_action_routes_use_success_envelope_and_validate_body(monkeypatch) -> No assert invalid.status_code == 422 +def test_write_routes_require_idempotency_key(monkeypatch) -> None: + """docs/05 §5.1：业务写接口必须携带 `Idempotency-Key`，缺失或过短都是 422。""" + with authenticated_client(monkeypatch, idempotent=False) as client: + missing = client.post("/api/v1/risk/alerts/ALERT-001/acknowledgements") + too_short = client.post( + "/api/v1/risk/alerts/ALERT-001/acknowledgements", + headers={"Idempotency-Key": "too-short"}, + ) + + assert missing.status_code == 422 + assert missing.json()["error"]["code"] == "AGENT_INPUT_INVALID" + assert too_short.status_code == 422 + assert too_short.json()["error"]["code"] == "AGENT_INPUT_INVALID" + + def test_evidence_upload_uses_success_envelope(monkeypatch) -> None: with authenticated_client(monkeypatch) as client: response = client.post(