fix: report capture's dropped duplicate copies as drops

.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.

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.

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,13 +607,21 @@ 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:
                     # Absent from the map: treated as accepted (matches
                     # 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)

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(