fix: ingest 同批去重/会话族切分修复/xdb 无关测试/边界加固

Co-Authored-By: Claude Code <noreply@anthropic.com>

server/tracker/services/ingest.py

@@ -1,17 +1,22 @@
 """批量入库服务：访客/会话 get_or_create、事件幂等写入、聚合更新。
 
 并发不变量（四轮审查收口）：
-  ① event 绝不重复入库：uq_site_event + bulk_create(ignore_conflicts=True) + 入库前 existed 预查；
+  ① event 绝不重复入库：uq_site_event + bulk_create(ignore_conflicts=True) + 入库前 existed 预查
+     + 同批 event_id 去重（seen_eids）；
   ② 重放命中的事件不参与会话聚合（new_events 只含未入库事件）；
   ③ visit_count/active_days/event_count/duration_ms/stay_ms 全走 F 表达式，
      filter().update() 避免 read-modify-write 全量 save() 用旧值覆盖并发赢家；
   ④ 会话 ended_at 用 Greatest 只前进不回退，乱序批次不改早结束时间；
   ⑤ 同批切分出的会话段按"原始 sid"缓存复用，后续事件不再切分。
+
+已接受的残余风险（M1 不修）：existed 预查与 bulk_create 之间存在毫秒级 TOCTOU 窗口——
+并发请求在预查后、写入前提交同一 event_id 时，本请求的 ignore_conflicts 静默吞掉冲突，
+但该事件已被计入 new_events 参与聚合，聚合可能双计。窗口极窄且仅影响并发重放场景。
 """
 from datetime import timedelta
 
 from django.db import transaction, IntegrityError
-from django.db.models import F
+from django.db.models import F, Q
 from django.db.models.functions import Greatest
 from django.utils import timezone
 
@@ -56,19 +61,24 @@ def _touch_visitor(site, vid, ctx, ip):
                        province=_geo_field(g, "province")[:64],
                        city=_geo_field(g, "city")[:64])
     Visitor.objects.filter(pk=v.pk).update(**upd)
-    v.refresh_from_db()
-    return v
+    return v   # 下游只用 pk（FK/子查询），无需 refresh_from_db 多耗一条 SELECT
 
 
 def _get_session(site, visitor, sid, ctx, occurred_at):
-    """会话权威在 SDK；服务端兜底：同 session_id 距上事件>30min 切分后缀段。"""
+    """会话权威在 SDK；服务端兜底：同 session_id 族（s1、s1#2…）距最后事件>30min 切分后缀段。
+
+    空闲比较必须取整个会话族的最大 ended_at：若只查根段（session_id=sid 精确匹配），
+    切分一次后根段 ended_at 陈旧，后续小间隔批次会反复误判超时、每批新开段（visit_count 灌倍）。
+    """
     idle = timedelta(milliseconds=C.SESSION_IDLE_MS)
-    sess = VisitSession.objects.filter(site=site, session_id=sid).order_by("-started_at").first()
-    if sess and occurred_at - sess.ended_at > idle:
-        suffix = 2 + VisitSession.objects.filter(
-            site=site, session_id__startswith=sid + "#").count()
-        sid = f"{sid}#{suffix}"
-        sess = None
+    fam = list(VisitSession.objects.filter(
+        Q(session_id=sid) | Q(session_id__startswith=sid + "#"), site=site))
+    last = max(fam, key=lambda s: s.ended_at) if fam else None
+    if last and occurred_at - last.ended_at > idle:
+        suffix = 2 + sum(1 for s in fam if s.session_id.startswith(sid + "#"))
+        sid, sess = f"{sid}#{suffix}", None
+    else:
+        sess = last
     if not sess:
         try:
             with transaction.atomic():   # savepoint：IntegrityError 后外层事务还能继续
@@ -92,6 +102,9 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di
     ctx = payload.get("context") or {}
     events = payload.get("events") or []
     result = {"accepted": 0, "duplicated": 0, "rejected": []}
+    if not payload.get("visitor_id") or not payload.get("session_id"):
+        return {"accepted": 0, "duplicated": 0,
+                "rejected": [{"reason": "invalid_payload"}]}
     if not events or len(events) > C.MAX_EVENTS_PER_BATCH:
         return {"accepted": 0, "duplicated": 0,
                 "rejected": [{"reason": "batch_size_invalid"}]}
@@ -100,13 +113,17 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di
     is_bot = any(k in (ua or "").lower() for k in C.BOT_UA_KEYWORDS)
     if is_bot:
         Visitor.objects.filter(pk=visitor.pk).update(is_bot=True)
-    ready, session_cache = [], {}
+    ready, session_cache, seen_eids = [], {}, set()
     for i, ev in enumerate(events):
         reason = validate_event(ev)
         if reason:
             result["rejected"].append({"index": i, "event_id": ev.get("event_id", ""),
                                        "reason": reason})
             continue
+        if ev["event_id"] in seen_eids:   # 同批重复 event_id：只入一份，防聚合/计数灌倍
+            result["duplicated"] += 1
+            continue
+        seen_eids.add(ev["event_id"])
         occ = ev["occurred_at"]
         sid = payload["session_id"]
         sess = session_cache.get(sid)
@@ -121,8 +138,8 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di
             title=(ev.get("title") or "")[:200],
             properties=ev.get("properties") or {},
             ip=ip, late=abs((now - occ).total_seconds()) > C.CLOCK_SKEW_LATE_SEC,
-            country=_geo_field(geo, "country"), province=_geo_field(geo, "province"),
-            city=_geo_field(geo, "city"), is_bot=is_bot))
+            country=_geo_field(geo, "country")[:64], province=_geo_field(geo, "province")[:64],
+            city=_geo_field(geo, "city")[:64], is_bot=is_bot))
     if ready:
         existed = set(Event.objects.filter(
             site=site, event_id__in=[e.event_id for e in ready]
@@ -130,7 +147,7 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di
         new_events = [e for e in ready if e.event_id not in existed]   # 重放命中的不参与聚合！
         Event.objects.bulk_create(new_events, ignore_conflicts=True)
         result["accepted"] = len(new_events)
-        result["duplicated"] = len(ready) - len(new_events)
+        result["duplicated"] += len(ready) - len(new_events)   # += ：累加同批去重已计入的份数
         for sess in {e.session for e in new_events}:
             se = [e for e in new_events if e.session.pk == sess.pk]
             new_end = max(x.occurred_at for x in se)

server/tracker/tests/test_ingest.py

@@ -1,6 +1,7 @@
 from datetime import timedelta
 from django.utils import timezone
 from tracker.models import Site, Visitor, VisitSession, Event
+from tracker.services.geo import resolver as _geo
 from tracker.services.ingest import ingest_batch
 
 NOW = timezone.now()
@@ -26,7 +27,8 @@ def test_first_batch_creates_visitor_session_events(db):
     v = Visitor.objects.get(site=site, visitor_id="v1")
     assert v.first_referrer_domain == "google.com" and v.visit_count == 1
     assert v.active_days == 1 and v.last_ip == "114.114.114.114"
-    assert v.country == "中国" and v.city == "南京市"    # 真实 xdb（114 DNS，江苏南京）；无 xdb 环境改为断言不抛异常即可
+    if _geo._searcher:    # 地理断言仅在有真实 xdb 时执行，CI/fresh clone 不因缺库变红
+        assert v.country == "中国" and v.city == "南京市"    # 114 DNS（江苏南京）
     assert Event.objects.filter(site=site).count() == 2
 
 # 2. 重放幂等 + 不抬高会话聚合
@@ -83,3 +85,52 @@ def test_batch_size_invalid(db):
     site = Site.objects.create(name="s", domains=["a.com"])
     r = ingest_batch(site, _batch([_ev(i) for i in range(60)]))
     assert r["rejected"] == [{"reason": "batch_size_invalid"}]
+
+# 9. 同批重复 event_id：去重后只入一份，聚合不灌倍
+def test_intra_batch_duplicate_event_id_deduped(db):
+    site = Site.objects.create(name="s", domains=["a.com"])
+    evs = [_ev(1, "page_stay", properties={"active_ms": 1000, "path": "/p"})
+           for _ in range(3)]                            # 3 份同 id
+    r = ingest_batch(site, _batch(evs))
+    assert (r["accepted"], r["duplicated"]) == (1, 2)
+    sess = VisitSession.objects.get()
+    assert sess.event_count == 1 and sess.stay_ms == 1000
+
+# 10. 切分后同 sid 小间隔连发：空闲比较取会话族最大 ended_at，不再每批新开段
+def test_split_then_small_gap_reuses_segment(db):
+    site = Site.objects.create(name="s", domains=["a.com"])
+    ingest_batch(site, _batch([_ev(1, occurred_at=NOW - timedelta(minutes=90))]))
+    ingest_batch(site, _batch([_ev(2)]))                 # 90min 间隔 → 切出 s1#2
+    ingest_batch(site, _batch([_ev(3, occurred_at=NOW + timedelta(minutes=1))]))
+    ingest_batch(site, _batch([_ev(4, occurred_at=NOW + timedelta(minutes=2))]))
+    assert VisitSession.objects.count() == 2             # 保持 s1 与 s1#2，不切 s1#3/s1#4
+    assert Visitor.objects.get(visitor_id="v1").visit_count == 2
+
+# 11. invariant 4：乱序批次（后入更早发生）ended_at 不回退
+def test_ended_at_never_regresses_on_out_of_order_batch(db):
+    site = Site.objects.create(name="s", domains=["a.com"])
+    ingest_batch(site, _batch([_ev(1, occurred_at=NOW + timedelta(minutes=1))]))
+    ingest_batch(site, _batch([_ev(2, occurred_at=NOW)]))
+    sess = VisitSession.objects.get()
+    assert abs((sess.ended_at - (NOW + timedelta(minutes=1))).total_seconds()) < 1
+
+# 12. 首触持久：第二批换 referrer 不改首触字段
+def test_first_touch_immutable_across_batches(db):
+    site = Site.objects.create(name="s", domains=["a.com"])
+    ingest_batch(site, _batch([_ev(1)]))
+    b = _batch([_ev(2)])
+    b["context"]["referrer_domain"] = "bing.com"
+    ingest_batch(site, b)
+    v = Visitor.objects.get(visitor_id="v1")
+    assert v.first_referrer_domain == "google.com"
+    assert v.first_utm == {"utm_source": "google"}
+
+# 13. payload 缺 visitor_id/session_id 提前拒绝
+def test_missing_payload_fields_rejected(db):
+    site = Site.objects.create(name="s", domains=["a.com"])
+    p = _batch([_ev(1)])
+    del p["visitor_id"]
+    assert ingest_batch(site, p)["rejected"] == [{"reason": "invalid_payload"}]
+    p = _batch([_ev(1)])
+    del p["session_id"]
+    assert ingest_batch(site, p)["rejected"] == [{"reason": "invalid_payload"}]