fix(alerts): deterministic tiebreaker for paginated alert list

list_alerts ordered by created_at DESC alone; kpi_alerts has many rows sharing the same second timestamp (07:00 batches), so MySQL tie order is unstable and offset pagination both duplicated and skipped rows (entity 1 read-only probe: total=281, 281 rows walked, only 268 unique ids, 13 duplicates; id 587 unreachable in the unfiltered list while visible in ?status=pending).

Add secondary key id DESC so pages form a total order; add regression suite test_alerts_pagination_stability_20260925.py (tie-heavy fixture: unique==total, zero duplicates, strict id DESC inside a tie group, page concat == single-shot ordered query, unfiltered vs status-filtered visibility, entity isolation negative control).
This commit is contained in:
Hermes CI Fix
2026-09-25 14:06:38 +08:00
parent a8391a03da
commit e3695088a4
2 changed files with 176 additions and 1 deletions
+7 -1
View File
@@ -23,7 +23,13 @@ def list_alerts(status: str = None, page: int = Query(1, ge=1), db: Session = De
if status: if status:
query = query.filter(KPIAlert.status == status) query = query.filter(KPIAlert.status == status)
total = query.count() total = query.count()
alerts = query.order_by(KPIAlert.created_at.desc()).offset((page-1)*20).limit(20).all() # 分页确定性(2026-09-25 alerts-list-pagination-skip):原为 created_at.desc() **单键** +
# offset/limit — kpi_alerts 每批 07:00 生成、**大量行同秒 created_at**,MySQL 并列组内顺序
# 不确定 ⇒ 偏移分页重复并漏行(entity 1 只读实测:total=281 / 逐页取满 281 行 / 唯一 id 仅 268 /
# 重复 13;id=587 在无过滤列表不可达,而同行 ?status=pending 首屏正常可见)。
# 补**确定性次键** id.desc()(同秒内后写入者在前)→ 全序分页,逐页遍历可覆盖全部 total 行。
alerts = (query.order_by(KPIAlert.created_at.desc(), KPIAlert.id.desc())
.offset((page-1)*20).limit(20).all())
return {"total": total, "data": [{c.name: getattr(a, c.name) for c in KPIAlert.__table__.columns} for a in alerts]} return {"total": total, "data": [{c.name: getattr(a, c.name) for c in KPIAlert.__table__.columns} for a in alerts]}
@router.post("/{alert_id}/resolve") @router.post("/{alert_id}/resolve")
@@ -0,0 +1,169 @@
# -*- coding: utf-8 -*-
"""`GET /api/cma/alerts` 无过滤列表分页可及性回归(任务 fullstack-20260925-alerts-list-pagination-skip)
缺陷(改前既有,2026-09-25 生产只读实测):`app/api/alerts.py::list_alerts` 用
`ORDER BY created_at DESC` **单键** + `offset/limit` 分页;`kpi_alerts` 每批 07:00 生成,
**大量行同秒 created_at** ⇒ MySQL filesort 在并列组内顺序不确定,偏移分页会**重复并漏行**
(entity 1 实测:total=281、逐页取满 281 行、唯一 id 仅 **268**、重复 13;id=587 因此
不出现在无过滤列表,而同行的 `?status=pending` 列表首屏正常可见)。
修复:排序补**确定性次键** `id DESC` ⇒ 同一 `created_at` 内按 id 降序(= 同秒内后写入者在前),
偏移分页成为全序分页,逐页遍历可覆盖全部 total 行且零重复。
本套件判据(改前 RED / 改后 GREEN):
· T1 唯一性:逐页取满后 `唯一 id 数 == total` 且 `重复 == 0`;
· T2 确定性次键:并列 `created_at` 组内 id **严格降序**(改前同一夹具下为插入序 = 升序 → RED);
· T3 逐页序列 == 「一次性全量取」(无分页)按 (created_at desc, id desc) 的序列,逐项相同;
· T4 可见性一致:同一行在「无过滤遍历」与「?status=pending 遍历」中一致可见(587 型);
· T5 负对照(非回归):账套隔离未放宽 —— 他账套行不出现在本账套列表;跨时间戳仍「新→旧」。
"""
from datetime import datetime, timedelta
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models import KPIAlert
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
PAGE_SIZE = 20 # app/api/alerts.py 硬编码 limit(20)
#: 固定「同秒」时间戳 —— 全部并列行共用,复刻生产「每批 07:00 生成、同秒 created_at」形态
SAME_SECOND = datetime(2026, 9, 20, 7, 0, 42)
def _alert(db: Session, kpi_id: int, **kw) -> KPIAlert:
defaults = {
"kpi_id": kpi_id,
"entity_id": 1,
"alert_level": "yellow",
"alert_message": "分页夹具",
"status": "pending",
"created_at": SAME_SECOND,
}
defaults.update(kw)
a = KPIAlert(**defaults)
db.add(a)
db.commit()
db.refresh(a)
return a
def _walk(client: TestClient, token: str, status: str = None) -> dict:
"""逐页调用真实端点,直到某页不足 PAGE_SIZE 行(= 前端/调用方遍历姿势)。"""
url = "/api/cma/alerts" + (f"?status={status}" if status else "")
pages, seen, dups, rows = [], [], [], []
page = 1
total = None
while True:
sep = "&" if "?" in url else "?"
resp = client.get(f"{url}{sep}page={page}", headers=auth_header(token))
assert resp.status_code == 200, resp.text
body = resp.json()
if total is None:
total = body["total"]
ids = [r["id"] for r in body["data"]]
for pos, r in enumerate(body["data"]):
if r["id"] in seen:
dups.append({"id": r["id"], "page": page, "pos": pos})
else:
seen.append(r["id"])
rows.append(r)
pages.append(ids)
if len(ids) < PAGE_SIZE:
break
page += 1
assert page < 100, "分页遍历未终止(保险丝)"
return {"total": total, "pages": pages, "unique": seen, "dups": dups, "rows": rows}
def _tie_groups(rows):
groups = {}
for r in rows:
groups.setdefault(str(r["created_at"]), []).append(r["id"])
return groups
class TestAlertsListPaginationAccessibility:
"""分页可及性(判据主体;改前 T1/T2/T3 应为 RED)"""
def test_t1_t2_t3_tie_heavy_batch_all_rows_reachable_and_deterministic(
self, client: TestClient, db: Session):
"""61 条同秒 + 5 条跨时间戳:逐页取满 ⇒ 唯一==total、零重复、并列组 id 降序、与全量序列一致"""
create_test_user(db)
token = get_token_for_user(client)
kpi = create_test_kpi(db)
# 61 条同秒(>3 页,且每页必然出现「并列组跨界」)+ 5 条更早的(跨时间戳排序正对照)
tie_ids = [_alert(db, kpi.id).id for _ in range(61)]
older_ids = [_alert(db, kpi.id, created_at=SAME_SECOND - timedelta(hours=1)).id for _ in range(5)]
total_expected = 66
w = _walk(client, token, None)
assert w["total"] == total_expected, w["total"]
# T1 唯一性与重复
assert len(set(w["unique"])) == w["total"], (
f"逐页遍历唯一 id 数 {len(set(w['unique']))} != total {w['total']};重复 {w['dups']}")
assert w["dups"] == [], f"偏移分页出现重复行:{w['dups']}"
assert len(w["unique"]) == w["total"], "逐页取到的行数应等于 total"
# T2 确定性次键:并列组内 id 严格降序
for ca, ids in _tie_groups(w["rows"]).items():
if len(ids) > 1:
assert ids == sorted(ids, reverse=True), (
f"created_at={ca} 并列组内未按 id 降序(分页顺序不确定):{ids}")
# T3 逐页拼接序列 == 一次性全量取(不同查询路径:单次全量 ORM 查询,显式两键排序)
full = (db.query(KPIAlert)
.filter(KPIAlert.entity_id == 1)
.order_by(KPIAlert.created_at.desc(), KPIAlert.id.desc())
.all())
assert [r["id"] for r in w["rows"]] == [a.id for a in full], "逐页序列与全量排序序列不一致"
# 全量序列 = 5 条跨时间戳(新→旧)+ 61 条同秒(id 降序)
assert [r["id"] for r in w["rows"]][:5] == sorted(tie_ids, reverse=True)[:5]
assert set(w["unique"]) == set(tie_ids) | set(older_ids)
def test_t4_visibility_consistent_between_unfiltered_and_status_filtered(
self, client: TestClient, db: Session):
"""同一行在无过滤遍历与 ?status=pending 遍历中一致可见(587 型:待办不得翻页翻不到)"""
create_test_user(db)
token = get_token_for_user(client)
kpi = create_test_kpi(db)
# 65 条同秒 pending(>3 页)+ 3 条同秒 resolved(无过滤列表里夹在并列组中)
pending_ids = [_alert(db, kpi.id, status="pending").id for _ in range(65)]
_alert_ids = [_alert(db, kpi.id, status="resolved").id for _ in range(3)]
oldest_pending = min(pending_ids) # 深页行:改前最易被漏读
unf = _walk(client, token, None)
pend = _walk(client, token, "pending")
assert unf["total"] == 68 and pend["total"] == 65
assert len(set(unf["unique"])) == unf["total"], "无过滤列表漏读"
assert len(set(pend["unique"])) == pend["total"], "pending 列表漏读"
assert oldest_pending in unf["unique"], "深页 pending 行在无过滤列表不可达(改前 RED 形态)"
assert oldest_pending in pend["unique"], "深页 pending 行在 pending 列表不可达"
for aid in pending_ids:
assert aid in unf["unique"], f"pending 行 {aid} 在无过滤列表不可达"
assert set(unf["unique"]) == set(pending_ids) | set(_alert_ids)
def test_t5_entity_isolation_and_cross_timestamp_order_unchanged(
self, client: TestClient, db: Session):
"""负对照:账套隔离未放宽(他账套行不可见)+ 跨时间戳仍「新→旧」"""
create_test_user(db)
token = get_token_for_user(client)
kpi = create_test_kpi(db)
mine = [_alert(db, kpi.id, created_at=SAME_SECOND - timedelta(days=2)).id for _ in range(30)]
other = [_alert(db, kpi.id, entity_id=2, created_at=SAME_SECOND).id for _ in range(30)]
# 本账套另有 6 条更新的行(**时间戳互不相同** → 无并列,纯跨时间戳正对照):
# 必须排在他账套「同秒组」之前,且按新→旧;改前改后都应通过(本用例是非回归负对照)
newer = [_alert(db, kpi.id,
created_at=SAME_SECOND + timedelta(hours=i + 1)).id for i in range(6)]
newer_expected = list(reversed(newer)) # 最新的一条(+6h)在最前
w = _walk(client, token, None)
assert w["total"] == len(mine) + len(newer), "账套隔离被放宽:他账套行进入本账套列表"
assert not (set(other) & set(w["unique"])), "他账套行泄漏进本账套列表"
assert [r["id"] for r in w["rows"]][:6] == newer_expected, "跨时间戳未按新→旧排序"
assert set(w["unique"]) == set(mine) | set(newer)