backend/app/repositories/rating_repository.py

@@ -11,14 +11,21 @@ from app.models import CustomerRatingGrade
 # ==================== 读取 ====================
 
 
-def list_all(session: Session) -> list[CustomerRatingGrade]:
+def list_all(session: Session, *, for_update: bool = False) -> list[CustomerRatingGrade]:
     """全量等级，按 sort_order 升序、id 兜底（同序次序稳定）。
 
     不分页：等级映射是判定表，数量个位数，分页无意义。
+
+    for_update 供区间重叠判定使用：先查后写在无锁时，两管理员并发提交
+    重叠区间会互判不重叠而双双入库（区间重叠无唯一键可拦）——锁全表行
+    串行化并发的重叠判定（先到者提交后，后到者在锁下读到新区间即判重叠）。
+    只读直出（list_grades）不加锁。
     """
     stmt = select(CustomerRatingGrade).order_by(
         CustomerRatingGrade.sort_order, CustomerRatingGrade.id
     )
+    if for_update:
+        stmt = stmt.with_for_update()
     return list(session.execute(stmt).scalars())
 
 

backend/app/services/rating_service.py

@@ -156,7 +156,7 @@ def create_grade(
         new_min = _to_decimal(payload.min_score)
         new_max = _to_decimal(payload.max_score)
         _ensure_score_range(new_min, new_max)
-        _ensure_no_overlap(rating_repository.list_all(session), new_min, new_max)
+        _ensure_no_overlap(rating_repository.list_all(session, for_update=True), new_min, new_max)
         if rating_repository.get_by_name(session, payload.grade_name) is not None:
             raise BadRequestError("该等级名称已存在")
 
@@ -217,7 +217,7 @@ def update_grade(
             raise NotFoundError("评级等级不存在")
 
         _ensure_no_overlap(
-            rating_repository.list_all(session), new_min, new_max, exclude_id=grade_id
+            rating_repository.list_all(session, for_update=True), new_min, new_max, exclude_id=grade_id
         )
         existed = rating_repository.get_by_name(session, payload.grade_name)
         if existed is not None and existed.id != grade_id:

backend/tests/integration/test_ratings.py

@@ -98,6 +98,27 @@ class TestAuth:
         assert resp.status_code == 401, resp.text
 
 
+class TestOverlapGuardLock:
+    """区间重叠判定的并发防线：判重叠的 list_all 必须带 FOR UPDATE。
+
+    区间重叠无唯一键可拦（名称才有 uk），无锁时两管理员并发提交重叠区间
+    会互判不重叠而双双入库。双连接真并发需线程与锁等待控制，此处锁定
+    SQL 形态两条口径：for_update 编译产物含 FOR UPDATE、只读直出不含。
+    """
+
+    def test_判重叠查询带FOR_UPDATE_只读直出不带(self) -> None:
+        from app.models import CustomerRatingGrade
+
+        locked = (
+            select(CustomerRatingGrade)
+            .order_by(CustomerRatingGrade.sort_order, CustomerRatingGrade.id)
+            .with_for_update()
+        )
+        assert "FOR UPDATE" in str(locked.compile())
+        plain = select(CustomerRatingGrade).order_by(CustomerRatingGrade.sort_order)
+        assert "FOR UPDATE" not in str(plain.compile())
+
+
 class TestList:
     """列表直出种子基线：判定表是前端评级计算器的唯一数据源。"""