fix: L3 乐观锁防丢更新/last_alert_id 保留语义/竞态日志(B3 评审 P1-1、P2-1~5)

app/repository/risk_repository.py

@@ -310,29 +310,33 @@ class RiskRepository:
         monitor_tags: list[str],
         last_alert_id: str | None,
         computed_at: datetime,
-    ) -> None:
-        """整行更新（service 层完成最高档/tags 合并后调用；computed_at NOT NULL 必传）。"""
-        sql = text(
-            """
+        expected_computed_at: datetime | None = None,
+    ) -> bool:
+        """整行更新（service 层完成最高档/tags 合并后调用；computed_at NOT NULL 必传）。
+
+        expected_computed_at 传入时为乐观锁（比对读时的 computed_at），行已被并发
+        修改则不命中，返回 False 由 service 层重读重试（B3 评审 P1-1 防丢更新）。
+        """
+        sql = """
             UPDATE customer_profile_l3
             SET monitor_tier = :tier, risk_score = :score, score_dimensions = :dims,
                 monitor_tags = :tags, last_alert_id = :last_alert, computed_at = :computed_at
-            WHERE customer_id = :cid
-            """
-        )
+            WHERE customer_id = :cid"""
+        params: dict[str, Any] = {
+            "cid": customer_id,
+            "tier": monitor_tier,
+            "score": risk_score,
+            "dims": json.dumps(score_dimensions or {}, ensure_ascii=False),
+            "tags": json.dumps(monitor_tags, ensure_ascii=False),
+            "last_alert": last_alert_id,
+            "computed_at": computed_at,
+        }
+        if expected_computed_at is not None:
+            sql += " AND computed_at = :expected"
+            params["expected"] = expected_computed_at
         with self._engine.begin() as conn:
-            conn.execute(
-                sql,
-                {
-                    "cid": customer_id,
-                    "tier": monitor_tier,
-                    "score": risk_score,
-                    "dims": json.dumps(score_dimensions or {}, ensure_ascii=False),
-                    "tags": json.dumps(monitor_tags, ensure_ascii=False),
-                    "last_alert": last_alert_id,
-                    "computed_at": computed_at,
-                },
-            )
+            res = conn.execute(text(sql), params)
+            return res.rowcount == 1
 
     # ---------- risk_aml_list ----------
 

app/service/risk/profile_l3.py

@@ -1,11 +1,15 @@
 """L3 监测画像写入（B3 · PRD FR-7 / R-05 最小写入）。
 
 合并规则（防降级，PRD FR-7）：monitor_tier 取最高档（normal < watch < high），
-monitor_tags 追加合并不覆盖，risk_score 取 max，computed_at 每次写当前时间
-（列 NOT NULL 必须显式赋值），last_alert_id 联动最新预警。
+monitor_tags 追加合并不覆盖，risk_score 取 max（FR-7 未定义的口径外推，见
+RISK-00n 静态分；R-05 动态评分接入时复核），last_alert_id 传入时联动最新预警
+（未传保留旧值，防调用方漏传抹掉），computed_at 每次写当前时间（列 NOT NULL，
+毫秒截断对齐 DATETIME(3)，保证乐观锁读写比对一致）。
 
-并发：进程内锁按 customer_id 串行（多进程部署换 Redis SET NX，接口不变）；
-锁超时降级与跨进程竞态由 IntegrityError 兜底——重读合并后再 update，不丢更新。
+并发（B3 评审 P1-1 修复）：进程内锁按 customer_id 串行减少冲突（多进程部署换
+Redis SET NX，接口不变）；锁超时降级与跨进程竞态由乐观锁兜底——update 比对
+读时 computed_at，未命中或 insert 撞主键则重读合并重试，3 次仍冲突抛错
+（风控数据宁失败不静默覆盖）。
 """
 
 from __future__ import annotations
@@ -30,6 +34,12 @@ ALERT_TYPE_TIER = {
     "suitability": "normal",
 }
 AML_PENDING_TAG = "aml_hit_pending_review"
+_MAX_RETRIES = 3
+
+
+def _ms(dt: datetime) -> datetime:
+    """毫秒截断（对齐表列 DATETIME(3)；否则乐观锁 expected 比对永不命中）。"""
+    return dt.replace(microsecond=(dt.microsecond // 1000) * 1000)
 
 
 def tier_of(alert_type: str) -> str:
@@ -60,12 +70,13 @@ def merge_l3(
     existing 为 None 表示新客户首写；risk_score 双方均空时保持 NULL。
     """
     if existing is None:
-        tier, old_score, old_tags, old_dims = "normal", None, [], {}
+        tier, old_score, old_tags, old_dims, old_alert = "normal", None, [], {}, None
     else:
         tier = existing.get("monitor_tier") or "normal"
         old_score = existing.get("risk_score")
         old_tags = existing.get("monitor_tags") or []
         old_dims = existing.get("score_dimensions") or {}
+        old_alert = existing.get("last_alert_id")
 
     scores = [int(s) for s in (old_score, risk_score) if s is not None]
     return {
@@ -73,8 +84,8 @@ def merge_l3(
         "risk_score": max(scores) if scores else None,
         "monitor_tags": sorted(set(old_tags) | set(monitor_tags)),
         "score_dimensions": score_dimensions if score_dimensions is not None else old_dims,
-        "last_alert_id": last_alert_id,
-        "computed_at": computed_at or datetime.now(),
+        "last_alert_id": last_alert_id if last_alert_id is not None else old_alert,
+        "computed_at": _ms(computed_at or datetime.now()),
     }
 
 
@@ -88,9 +99,10 @@ def upsert_profile_l3(
     risk_repo: RiskRepository | None = None,
     computed_at: datetime | None = None,
 ) -> dict[str, Any]:
-    """预警事件 → L3 upsert（B2 预警落库后调用；aml 自动追加待复核标签）。
+    """预警事件 → L3 upsert（aml 自动追加待复核标签）。
 
-    返回合并后的 L3 行（含 customer_id）。
+    返回合并后的 L3 行（含 customer_id）。B7 接 Redis 缓存时须在此处加
+    写侧 DEL 钩子（PRD §5.1：MySQL 更新时 DEL profile:l3:{customer_id}）。
     """
     repo = risk_repo or RiskRepository()
     mapped_tier = tier_of(alert_type)
@@ -98,45 +110,11 @@ def upsert_profile_l3(
     if alert_type == "aml" and AML_PENDING_TAG not in tags:
         tags.append(AML_PENDING_TAG)
 
-    def _write(locked: bool) -> dict[str, Any]:
-        existing = repo.get_l3(customer_id)
-        merged = merge_l3(
-            existing,
-            mapped_tier=mapped_tier,
-            risk_score=risk_score,
-            monitor_tags=tags,
-            last_alert_id=last_alert_id,
-            score_dimensions=score_dimensions,
-            computed_at=computed_at,
-        )
-        try:
-            if existing is None:
-                repo.insert_l3(
-                    customer_id,
-                    merged["monitor_tier"],
-                    merged["risk_score"],
-                    merged["score_dimensions"],
-                    merged["monitor_tags"],
-                    merged["last_alert_id"],
-                    merged["computed_at"],
-                )
-            else:
-                repo.update_l3(
-                    customer_id,
-                    merged["monitor_tier"],
-                    merged["risk_score"],
-                    merged["score_dimensions"],
-                    merged["monitor_tags"],
-                    merged["last_alert_id"],
-                    merged["computed_at"],
-                )
-        except IntegrityError:
-            # 首写竞态（锁超时降级/跨进程）：另一线程已 insert，重读合并转更新
-            if existing is not None:
-                raise
-            raced = repo.get_l3(customer_id)
+    def _write(locked: bool) -> dict[str, Any]:  # 锁原语签名要求（是否获得锁）；冲突安全由乐观锁重试保证
+        for attempt in range(1, _MAX_RETRIES + 1):
+            existing = repo.get_l3(customer_id)
             merged = merge_l3(
-                raced,
+                existing,
                 mapped_tier=mapped_tier,
                 risk_score=risk_score,
                 monitor_tags=tags,
@@ -144,8 +122,7 @@ def upsert_profile_l3(
                 score_dimensions=score_dimensions,
                 computed_at=computed_at,
             )
-            repo.update_l3(
-                customer_id,
+            fields = (
                 merged["monitor_tier"],
                 merged["risk_score"],
                 merged["score_dimensions"],
@@ -153,12 +130,23 @@ def upsert_profile_l3(
                 merged["last_alert_id"],
                 merged["computed_at"],
             )
-        merged["customer_id"] = customer_id
-        return merged
+            if existing is None:
+                try:
+                    repo.insert_l3(customer_id, *fields)
+                    merged["customer_id"] = customer_id
+                    return merged
+                except IntegrityError:
+                    logger.warning("L3 insert race, retrying (attempt %d): %s", attempt, customer_id)
+                    continue
+            if repo.update_l3(customer_id, *fields, expected_computed_at=existing["computed_at"]):
+                merged["customer_id"] = customer_id
+                return merged
+            logger.warning("L3 update lost race, retrying (attempt %d): %s", attempt, customer_id)
+        raise RuntimeError(f"L3 upsert conflicted after {_MAX_RETRIES} retries: {customer_id}")
 
     return _run_locked(f"l3:{customer_id}", _write)
 
 
 def get_profile_l3(customer_id: str, risk_repo: RiskRepository | None = None) -> dict[str, Any] | None:
-    """L3 只读薄封装（对话线/引擎复用；Redis 缓存待 B7 lifespan 一并接入）。"""
+    """L3 只读薄封装（对话线/引擎复用；Redis 热缓存与写侧 DEL 钩子待 B7 lifespan 一并接入）。"""
     return (risk_repo or RiskRepository()).get_l3(customer_id)