diff --git a/.sampo/changesets/duplicate-event-uuid.md b/.sampo/changesets/duplicate-event-uuid.md new file mode 100644 index 00000000..dc318428 --- /dev/null +++ b/.sampo/changesets/duplicate-event-uuid.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: patch +--- + +Capture keeps the first event for each uuid in a request and drops the later copies. `on_error` now reports each dropped copy in `CaptureError.drops` as `(uuid, "duplicate_event_uuid")`, and the SDK does not resend it. `verdict_summary()` counts each drop by its own reason. diff --git a/docs/migration-7.x-to-8.0.md b/docs/migration-7.x-to-8.0.md index fccf1244..87abcce7 100644 --- a/docs/migration-7.x-to-8.0.md +++ b/docs/migration-7.x-to-8.0.md @@ -118,7 +118,7 @@ Capture rejects a whole batch when one event cannot be parsed. ## Event uuids -- Each event needs its own uuid. Capture rejects a whole batch that contains the same uuid twice, so the other events in that batch are lost too. +- Each event needs its own uuid. Capture keeps the first event for each uuid in a request and drops the later copies. `on_error` reports each dropped copy in `CaptureError.drops` as `(uuid, "duplicate_event_uuid")`. 7.x sent every copy. Older PostHog versions reject the whole batch with a 400 `duplicate_event_uuid` instead. - The SDK accepts a uuid with hyphens, 32 hex digits, `{...}` braces or a `urn:uuid:` prefix, in any case. It sends the lowercase hyphenated form and returns that form from `capture`. - Any other value is replaced with a generated uuid, and the SDK logs one warning that names the rule, not the value. An empty string counts as unset, so the SDK generates a uuid without a warning. This applies to `AsyncPosthog` too. - Generated uuids are UUIDv7. diff --git a/posthog/capture_send.py b/posthog/capture_send.py index 9310df46..90c55b54 100644 --- a/posthog/capture_send.py +++ b/posthog/capture_send.py @@ -13,7 +13,9 @@ events succeed; ``drop`` and retry-exhaustion are carried on the :class:`CaptureError` raised on batch-level/terminal failure, so the consumer's existing ``on_error(exc, batch)`` path surfaces them unchanged (no per-event -logging of its own). +logging of its own). Capture keeps the first event for each uuid in a request; +a result's ``duplicates_dropped`` count reports the later copies as drops with +reason ``duplicate_event_uuid``. Request bodies are optionally compressed per :class:`~posthog.capture_compression.CaptureCompression` (``gzip`` or zlib-wrapped ``deflate``), advertised via ``Content-Encoding``. @@ -24,7 +26,7 @@ import random import time import zlib -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime, timezone from email.utils import parsedate_to_datetime from gzip import GzipFile @@ -74,6 +76,9 @@ _RESULT_DROP = "drop" _RESULT_RETRY = "retry" +# Drop reason for a later copy of a uuid that capture discarded in favor of the first. +_DUPLICATE_EVENT_UUID = "duplicate_event_uuid" + # HTTP status classification. 429 is terminal in v1 (unlike v0, where it is # retried) — the backend signals overload via retryable 5xx + Retry-After. _RETRYABLE_STATUSES = frozenset({408, 500, 502, 503, 504}) @@ -118,12 +123,15 @@ class _V1ParsedResponse: per-uuid directives (``None``/``malformed=True`` when the body could not be parsed — treated as terminal so a bad success never loops forever). On a non-2xx, ``error_message`` is the best-effort human-readable detail. + ``duplicates_dropped`` maps a uuid to the number of its later copies + capture dropped; it is empty when no result carries the field. """ status_code: int is_success: bool retry_after: Optional[float] = None results: Optional[dict[str, CaptureEventResult]] = None + duplicates_dropped: dict[str, int] = field(default_factory=dict) malformed: bool = False error_message: str = "" @@ -177,15 +185,19 @@ def verdict_summary(self) -> str: Empty when no event is counted. """ counts: dict[str, int] = {} - failed = [uid for uid, _ in self.drops] + self.retry_exhausted - for uid in failed: - verdict = self.event_results.get(uid) - if verdict is None or not verdict.result: - continue - tag = verdict.result - if verdict.details: - tag = f"{tag}/{verdict.details}" + + def count(result: str, details: Optional[str]) -> None: + tag = f"{result}/{details}" if details else result counts[tag] = counts.get(tag, 0) + 1 + + # A dropped duplicate shares its uuid with the kept copy, so each drop + # counts its own reason rather than the uuid's verdict. + for _, details in self.drops: + count(_RESULT_DROP, details) + for uid in self.retry_exhausted: + verdict = self.event_results.get(uid) + if verdict is not None and verdict.result: + count(verdict.result, verdict.details) return ", ".join(f"{tag}={n}" for tag, n in sorted(counts.items())) @@ -393,15 +405,23 @@ def _parse_v1_response(res: "requests.Response") -> _V1ParsedResponse: if _is_success_status(status): try: payload = res.json() - raw_results = payload["results"] - results = { - uid: CaptureEventResult( - result=(r or {}).get("result"), - details=(r or {}).get("details"), + results: dict[str, CaptureEventResult] = {} + duplicates_dropped: dict[str, int] = {} + for uid, r in payload["results"].items(): + r = r or {} + results[uid] = CaptureEventResult( + result=r.get("result"), details=r.get("details") ) - for uid, r in raw_results.items() - } - return _V1ParsedResponse(status, True, retry_after, results=results) + count = r.get("duplicates_dropped") + if isinstance(count, int) and not isinstance(count, bool) and count > 0: + duplicates_dropped[uid] = count + return _V1ParsedResponse( + status, + True, + retry_after, + results=results, + duplicates_dropped=duplicates_dropped, + ) except (ValueError, KeyError, AttributeError, TypeError): # 2xx with a body we can't read as a results map: terminal, so we # don't loop forever re-sending against a broken success. @@ -498,7 +518,10 @@ def _send_v1_batch( when the request itself was a 2xx (a success status is not full delivery) and even when a later attempt clears the outstanding retries — matching posthog-go (per-event failure callback) and posthog-rs (``on_error`` on a - 2xx with undelivered verdicts). Raises :class:`CaptureError` on any drop, + 2xx with undelivered verdicts). When a result carries + ``duplicates_dropped: n``, the first event with that uuid takes the verdict + and the next ``n`` copies become drops with reason ``duplicate_event_uuid``, + never resent. Raises :class:`CaptureError` on any drop, batch-level terminal failure, retry exhaustion, or transport failure (``status`` 0, chained to the transport error), carrying the endpoint, the accumulated ``drops``, any exhausted uuids, and every verdict seen so far, @@ -584,6 +607,8 @@ def capture_error( retry_events: list[dict] = [] retry_uuids: list[str] = [] + duplicates_left = dict(parsed.duplicates_dropped) + seen_duplicated: set[str] = set() for event, uid in zip(pending_events, pending_uuids): directive = results.get(uid) if directive is None: @@ -591,6 +616,12 @@ def capture_error( # posthog-go and posthog-rs). Capture answers every event # of a 2xx, so this only happens if it breaks that contract. continue + if uid in duplicates_left: + if uid in seen_duplicated and duplicates_left[uid] > 0: + duplicates_left[uid] -= 1 + all_drops.append((uid, _DUPLICATE_EVENT_UUID)) + continue + seen_duplicated.add(uid) event_results[uid] = directive if directive.result == _RESULT_RETRY: retry_events.append(event) diff --git a/posthog/test/test_capture_send.py b/posthog/test/test_capture_send.py index a2da43d7..2a951e8d 100644 --- a/posthog/test/test_capture_send.py +++ b/posthog/test/test_capture_send.py @@ -111,6 +111,9 @@ def __call__( return outcome +_DUP_A = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" + + def _msg(uuid, event="e", **overrides): msg = { "event": event, @@ -124,16 +127,18 @@ def _msg(uuid, event="e", **overrides): return msg -def _results_response(directives, headers=None): +def _results_response(directives, headers=None, duplicates_dropped=None): """200 response whose ``results`` map tags each uuid. ``directives`` maps uuid -> ``"ok"`` (or any result string) or a - ``(result, details)`` tuple. + ``(result, details)`` tuple. ``duplicates_dropped`` maps uuid -> count. """ results = {} for uid, spec in directives.items(): result, details = spec if isinstance(spec, tuple) else (spec, None) results[uid] = {"result": result, "details": details} + for uid, count in (duplicates_dropped or {}).items(): + results[uid]["duplicates_dropped"] = count return _FakeResponse(200, json_body={"results": results}, headers=headers) @@ -331,6 +336,31 @@ def test_success_parses_results_with_details(self) -> None: self.assertEqual(parsed.results["u-1"].result, "drop") self.assertEqual(parsed.results["u-1"].details, "spam") + def test_success_parses_duplicates_dropped(self) -> None: + res = _results_response( + {"u-1": "ok", "u-2": "ok"}, duplicates_dropped={"u-1": 2} + ) + parsed = _parse_v1_response(res) + self.assertEqual(parsed.duplicates_dropped, {"u-1": 2}) + self.assertEqual(parsed.results["u-1"], CaptureEventResult("ok")) + + @parameterized.expand( + [ + ("string", "2"), + ("bool", True), + ("negative", -1), + ("zero", 0), + ("float", 2.0), + ("null", None), + ] + ) + def test_invalid_duplicates_dropped_ignored(self, _name, value) -> None: + res = _results_response({"u-1": "ok"}, duplicates_dropped={"u-1": value}) + parsed = _parse_v1_response(res) + self.assertFalse(parsed.malformed) + self.assertEqual(parsed.duplicates_dropped, {}) + self.assertEqual(parsed.results["u-1"], CaptureEventResult("ok")) + @parameterized.expand( [ ("unparseable_body", _FakeResponse(200, raise_json=True)), @@ -550,6 +580,71 @@ def test_results_match_non_canonical_uuid(self, _name, sent_uuid) -> None: self.assertEqual(len(stub.calls), 2) self.assertEqual(exc.drops, [(canonical, "llm_events_over_quota")]) + @parameterized.expand( + [ + ("kept_ok", "ok", 2, [], [], [], CaptureEventResult("ok"), ""), + ( + "kept_warning", + "warning", + 2, + [], + [], + [], + CaptureEventResult("warning"), + "", + ), + ( + "kept_drop", + ("drop", "invalid_options"), + 2, + [], + [], + [(_DUP_A, "invalid_options")], + CaptureEventResult("drop", "invalid_options"), + ", drop/invalid_options=1", + ), + ( + "kept_retry_then_ok", + "retry", + 2, + [_results_response({_DUP_A: "ok"})], + [[_DUP_A]], + [], + CaptureEventResult("ok"), + "", + ), + ( + "copy_past_count_keeps_kept_verdict", + "retry", + 1, + [_results_response({_DUP_A: "ok"}, duplicates_dropped={_DUP_A: 1})], + [[_DUP_A, _DUP_A]], + [], + CaptureEventResult("ok"), + "", + ), + ] + ) + def test_duplicates_dropped_reported_as_drops( + self, _name, kept, count, later, resent, kept_drops, kept_result, summary_rest + ) -> None: + batch = [_msg(_DUP_A), _msg("u-b"), _msg(_DUP_A.upper()), _msg(_DUP_A)] + first = _results_response( + {_DUP_A: kept, "u-b": "ok"}, duplicates_dropped={_DUP_A: count} + ) + stub, exc = self._run_expecting_error(batch, [first, *later]) + + self.assertEqual([c["uuids"] for c in stub.calls[1:]], resent) + duplicate = (_DUP_A, "duplicate_event_uuid") + self.assertEqual(exc.drops, kept_drops + [duplicate, duplicate]) + self.assertEqual( + exc.event_results, + {_DUP_A: kept_result, "u-b": CaptureEventResult("ok")}, + ) + self.assertEqual( + exc.verdict_summary(), "drop/duplicate_event_uuid=2" + summary_rest + ) + def test_malformed_2xx_is_terminal(self) -> None: stub, exc = self._run_expecting_error( [_msg("u-1")], [_FakeResponse(200, raise_json=True)] @@ -667,6 +762,25 @@ class TestCaptureLossMessage(unittest.TestCase): "2 event(s) not persisted by /i/v1/analytics/events: 1 dropped, " "1 out of retries (drop/llm_events_over_quota=1, retry/not_persisted=1)", ), + ( + "duplicate_drop_counts_its_own_reason", + CaptureError( + 200, + "2 event(s) dropped by the server", + endpoint=_CAPTURE_V1_PATH, + drops=[ + ("u-a", "duplicate_event_uuid"), + ("u-drop", "llm_events_over_quota"), + ], + event_results={ + "u-a": CaptureEventResult("ok"), + "u-drop": CaptureEventResult("drop", "llm_events_over_quota"), + }, + ), + 3, + "2 event(s) not persisted by /i/v1/analytics/events: 2 dropped, " + "0 out of retries (drop/duplicate_event_uuid=1, drop/llm_events_over_quota=1)", + ), ( "request_failure_after_partial_success", CaptureError(