From 173d13dbf3e3aefb7db5170b149b667b2b13cd3b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 15 Sep 2026 12:59:13 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20ingest=20=E5=90=8C=E6=89=B9=E5=8E=BB?= =?UTF-8?q?=E9=87=8D/=E4=BC=9A=E8=AF=9D=E6=97=8F=E5=88=87=E5=88=86?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D/xdb=20=E6=97=A0=E5=85=B3=E6=B5=8B=E8=AF=95/?= =?UTF-8?q?=E8=BE=B9=E7=95=8C=E5=8A=A0=E5=9B=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- server/tracker/services/ingest.py | 47 +++++++++++++++++-------- server/tracker/tests/test_ingest.py | 53 ++++++++++++++++++++++++++++- 2 files changed, 84 insertions(+), 16 deletions(-) diff --git a/server/tracker/services/ingest.py b/server/tracker/services/ingest.py index 6e206df..a350a9c 100644 --- a/server/tracker/services/ingest.py +++ b/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) diff --git a/server/tracker/tests/test_ingest.py b/server/tracker/tests/test_ingest.py index 8e08dbc..3505d78 100644 --- a/server/tracker/tests/test_ingest.py +++ b/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"}]