fix(capture): stabilize v1 created_at, isolate compression, quiet drops

Address review of the v1 transport:
- Hoist the batch created_at out of the retry loop so the envelope stays
  stable across attempts (only the events list and PostHog-Attempt change).
- Isolate v1 request compression behind a CaptureCompression selector
  supporting gzip and zlib-wrapped deflate (RFC 1950), reverting the
  gzip_compress extraction from request.py so the v1 path owns its codecs.
- Stop logging per-event drops at WARNING; a server-chosen drop on a 2xx
  is not a delivery failure and is already carried on CaptureV1Error for
  batch-level surfacing via on_error.

Author: eli-r-ph

posthog/capture_compression.py

@@ -0,0 +1,90 @@
+import logging
+import os
+from enum import Enum
+from typing import Optional, Union
+
+log = logging.getLogger("posthog")
+
+CAPTURE_COMPRESSION_ENV_VAR = "POSTHOG_CAPTURE_COMPRESSION"
+
+
+class CaptureCompression(str, Enum):
+    """Selects the request-body compression for capture-v1 uploads.
+
+    Only honored when ``capture_mode`` is ``V1``; the legacy ``/batch/`` path
+    keeps using its own ``gzip`` flag. ``NONE`` sends the body uncompressed.
+    ``GZIP`` and ``DEFLATE`` (zlib, RFC 1950) are both stdlib / zero-dependency
+    and map to the matching ``Content-Encoding`` token the v1 server decodes
+    (``br``/``zstd`` are accepted by the server too but need extra dependencies,
+    so they are intentionally left out for now). Inheriting from ``str`` keeps
+    the members comparable to and serializable as their token values.
+    """
+
+    NONE = "none"
+    GZIP = "gzip"
+    DEFLATE = "deflate"
+
+
+# Accepted spellings for both the kwarg and the env var. ``identity`` mirrors
+# the HTTP token for "no encoding".
+_ALIASES: dict[str, CaptureCompression] = {
+    "none": CaptureCompression.NONE,
+    "identity": CaptureCompression.NONE,
+    "gzip": CaptureCompression.GZIP,
+    "deflate": CaptureCompression.DEFLATE,
+}
+
+
+def _coerce_explicit(
+    value: Union[CaptureCompression, str],
+) -> CaptureCompression:
+    """Normalize an explicitly-supplied compression to a ``CaptureCompression``.
+
+    An explicit but unrecognized value is a programming error, so it raises
+    ``ValueError`` rather than silently defaulting (unlike the env var, which is
+    operator-supplied and defaults defensively).
+    """
+    if isinstance(value, CaptureCompression):
+        return value
+    if isinstance(value, str):
+        resolved = _ALIASES.get(value.strip().lower())
+        if resolved is not None:
+            return resolved
+    raise ValueError(
+        f"invalid capture_compression {value!r}; expected a CaptureCompression "
+        f"or one of {sorted(_ALIASES)}"
+    )
+
+
+def resolve_capture_compression(
+    capture_compression: Optional[Union[CaptureCompression, str]] = None,
+    *,
+    gzip_fallback: bool = False,
+) -> CaptureCompression:
+    """Resolve the effective v1 compression.
+
+    Precedence: explicit ``capture_compression`` argument >
+    ``POSTHOG_CAPTURE_COMPRESSION`` env var > the legacy ``gzip`` flag
+    (``GZIP`` when set) > ``NONE``. An unrecognized env value logs a warning and
+    falls back to the ``gzip`` flag, so a typo never silently changes encoding.
+    """
+    if capture_compression is not None:
+        return _coerce_explicit(capture_compression)
+
+    fallback = CaptureCompression.GZIP if gzip_fallback else CaptureCompression.NONE
+
+    raw = os.environ.get(CAPTURE_COMPRESSION_ENV_VAR)
+    if raw is None or raw.strip() == "":
+        return fallback
+
+    resolved = _ALIASES.get(raw.strip().lower())
+    if resolved is None:
+        log.warning(
+            "Unrecognized %s=%r; falling back to %s. Expected one of %s.",
+            CAPTURE_COMPRESSION_ENV_VAR,
+            raw,
+            fallback.value,
+            sorted(_ALIASES),
+        )
+        return fallback
+    return resolved

posthog/capture_v1.py

@@ -21,28 +21,36 @@
 The response is per-event: a 200 carries a ``results`` map keyed by event uuid,
 each tagged ``ok``/``warning`` (terminal-success), ``drop`` (terminal-failure),
 or ``retry``. :func:`send_v1_batch` resends only the ``retry`` events on the next
-attempt (a stable ``PostHog-Request-Id`` and ``created_at`` across attempts; an
-incrementing ``PostHog-Attempt``), logs drops, and raises a
-:class:`CaptureV1Error` on batch-level/terminal failure or retry exhaustion so
-the consumer's existing ``on_error(exc, batch)`` path fires unchanged.
+attempt, holding the ``PostHog-Request-Id`` and batch ``created_at`` stable
+across attempts while incrementing ``PostHog-Attempt``. ``ok``/``warning``/absent
+events succeed; ``drop`` and retry-exhaustion are carried on the
+:class:`CaptureV1Error` 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).
+
+Request bodies are optionally compressed per :class:`~posthog.capture_compression.CaptureCompression`
+(``gzip`` or zlib-wrapped ``deflate``), advertised via ``Content-Encoding``.
 """
 
 import json
 import logging
 import time
+import zlib
 from collections.abc import Callable
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from email.utils import parsedate_to_datetime
-from typing import TYPE_CHECKING, Any, Optional, cast
+from gzip import GzipFile
+from io import BytesIO
+from typing import TYPE_CHECKING, Any, Optional
 from uuid import uuid4
 
+from posthog.capture_compression import CaptureCompression
 from posthog.request import (
     DatetimeSerializer,
     USER_AGENT,
     APIError,
     _get_session,
-    gzip_compress,
     normalize_host,
 )
 from posthog.utils import guess_timezone, remove_trailing_slash
@@ -197,14 +205,20 @@ def to_v1_event(msg: dict) -> dict:
     return event
 
 
-def build_v1_batch_body(events: list[dict], historical_migration: bool = False) -> dict:
+def build_v1_batch_body(
+    events: list[dict],
+    historical_migration: bool = False,
+    created_at: Optional[str] = None,
+) -> dict:
     """Assemble the v1 batch envelope.
 
     Carries no ``api_key`` (Bearer auth) and no ``sent_at``.
     ``historical_migration`` is omitted when False (the server defaults it).
+    ``created_at`` defaults to now in UTC; :func:`send_v1_batch` passes a value
+    hoisted once so it stays stable across retry attempts.
     """
     body: dict[str, Any] = {
-        "created_at": datetime.now(timezone.utc).isoformat(),
+        "created_at": created_at or datetime.now(timezone.utc).isoformat(),
         "batch": events,
     }
     if historical_migration:
@@ -285,14 +299,35 @@ def _parse_retry_after(header_value: Optional[str]) -> Optional[float]:
         return None
 
 
+def _compress_v1(
+    compression: CaptureCompression, data: str
+) -> tuple[str | bytes, Optional[str]]:
+    """Compress a v1 request body, returning ``(body, Content-Encoding token)``.
+
+    ``GZIP`` emits a gzip stream; ``DEFLATE`` emits a *zlib-wrapped* deflate
+    stream (RFC 1950, leading ``0x78``) to match posthog-go / posthog-rs and the
+    server's zlib decoder for ``Content-Encoding: deflate`` — raw, headerless
+    deflate would be misrouted. ``NONE`` returns the string body and no token.
+    """
+    if compression == CaptureCompression.GZIP:
+        buf = BytesIO()
+        with GzipFile(fileobj=buf, mode="w") as gz:
+            # `data` is produced by json.dumps(), whose default encoding is utf-8.
+            gz.write(data.encode("utf-8"))
+        return buf.getvalue(), "gzip"
+    if compression == CaptureCompression.DEFLATE:
+        return zlib.compress(data.encode("utf-8")), "deflate"
+    return data, None
+
+
 def post_v1(
     api_key: str,
     host: Optional[str],
     batch_body: dict,
     *,
     attempt: int,
     request_id: str,
-    gzip: bool = False,
+    compression: CaptureCompression = CaptureCompression.NONE,
     timeout: int = 15,
     session: Optional["requests.Session"] = None,
 ) -> "requests.Response":
@@ -301,11 +336,13 @@ def post_v1(
     Bearer-authed (no ``api_key`` in the body) with the required v1 headers.
     ``attempt`` (1-based) and the stable ``request_id`` are echoed via
     ``PostHog-Attempt``/``PostHog-Request-Id`` so the backend can correlate
-    retries. Returns the raw response; classification is left to the caller.
+    retries. The body is compressed per ``compression`` (advertised via
+    ``Content-Encoding``). Returns the raw response; classification is left to
+    the caller.
     """
     trimmed_host = remove_trailing_slash(normalize_host(host))
     url = trimmed_host + CAPTURE_V1_PATH
-    data: str | bytes = json.dumps(batch_body, cls=DatetimeSerializer)
+    data = json.dumps(batch_body, cls=DatetimeSerializer)
     headers = {
         "Content-Type": "application/json",
         "User-Agent": USER_AGENT,
@@ -315,13 +352,13 @@ def post_v1(
         HEADER_REQUEST_ID: request_id,
         HEADER_REQUEST_TIMESTAMP: datetime.now(timezone.utc).isoformat(),
     }
-    if gzip:
-        headers["Content-Encoding"] = "gzip"
-        data = gzip_compress(cast(str, data))
+    body, encoding = _compress_v1(compression, data)
+    if encoding is not None:
+        headers["Content-Encoding"] = encoding
 
     log.debug("capture v1 POST %s attempt=%s request_id=%s", url, attempt, request_id)
     return (session or _get_session()).post(
-        url, data=data, headers=headers, timeout=timeout
+        url, data=body, headers=headers, timeout=timeout
     )
 
 
@@ -404,7 +441,7 @@ def send_v1_batch(
     host: Optional[str],
     batch: list[dict],
     *,
-    gzip: bool = False,
+    compression: CaptureCompression = CaptureCompression.NONE,
     timeout: int = 15,
     max_retries: int = 3,
     historical_migration: bool = False,
@@ -414,21 +451,29 @@ def send_v1_batch(
 
     The v1 sibling of ``Consumer._send``: it loops up to ``max_retries + 1``
     attempts, but unlike v0 it shrinks the batch to only the events the server
-    tagged ``retry`` after each 2xx. ``ok``/``warning``/absent events succeed
-    silently; ``drop`` events are logged (a successful request that the server
-    chose to drop is not a delivery failure). Raises :class:`CaptureV1Error`
-    (or the underlying transport exception) on a batch-level terminal/transport
-    failure or once retries are exhausted, so the caller's ``on_error`` fires.
+    tagged ``retry`` after each 2xx. ``ok``/``warning``/absent events succeed; a
+    server-chosen ``drop`` on an otherwise-successful request is not a delivery
+    failure, so it is not raised or logged per-event (the DEBUG summary tallies
+    it). Raises :class:`CaptureV1Error` (or the underlying transport exception)
+    on a batch-level terminal/transport failure or once retries are exhausted —
+    carrying any ``drops`` and exhausted uuids — so the caller's ``on_error``
+    fires unchanged. ``request_id`` and the batch ``created_at`` are stable
+    across attempts; ``PostHog-Attempt`` increments.
     """
     request_id = str(uuid4())
+    # Hoisted once so the batch envelope is byte-identical across retry attempts
+    # (only the events list shrinks and the attempt header increments).
+    created_at = datetime.now(timezone.utc).isoformat()
     pending_events = [to_v1_event(m) for m in batch]
     pending_uuids = [e["uuid"] for e in pending_events]
     last_exc: Optional[Exception] = None
 
     for attempt_index in range(max_retries + 1):
         attempt = attempt_index + 1
         last_attempt = attempt_index == max_retries
-        body = build_v1_batch_body(pending_events, historical_migration)
+        body = build_v1_batch_body(
+            pending_events, historical_migration, created_at=created_at
+        )
 
         try:
             res = post_v1(
@@ -437,7 +482,7 @@ def send_v1_batch(
                 body,
                 attempt=attempt,
                 request_id=request_id,
-                gzip=gzip,
+                compression=compression,
                 timeout=timeout,
                 session=session,
             )
@@ -477,14 +522,6 @@ def send_v1_batch(
                     drops.append((uid, directive.details))
                 # ok / warning / unrecognized -> terminal success.
 
-            for uid, details in drops:
-                log.warning(
-                    "capture v1 dropped event uuid=%s request_id=%s details=%s",
-                    uid,
-                    request_id,
-                    details or "",
-                )
-
             if not retry_uuids:
                 return
             if last_attempt:

posthog/request.py

@@ -50,15 +50,6 @@ def _mask_tokens_in_url(url: str) -> str:
     return re.sub(r"(token=)([^&]{10})[^&]*", r"\1\2...", url)
 
 
-def gzip_compress(data: str) -> bytes:
-    """Gzip-compress a UTF-8 string for an ``Content-Encoding: gzip`` body."""
-    buf = BytesIO()
-    with GzipFile(fileobj=buf, mode="w") as gz:
-        # `data` is produced by json.dumps(), whose default encoding is utf-8.
-        gz.write(data.encode("utf-8"))
-    return buf.getvalue()
-
-
 @dataclass
 class GetResponse:
     """Response from a GET request with ETag support."""
@@ -251,7 +242,12 @@ def post(
     headers = {"Content-Type": "application/json", "User-Agent": USER_AGENT}
     if gzip:
         headers["Content-Encoding"] = "gzip"
-        data = gzip_compress(cast(str, data))
+        buf = BytesIO()
+        with GzipFile(fileobj=buf, mode="w") as gz:
+            # 'data' was produced by json.dumps(),
+            # whose default encoding is utf-8.
+            gz.write(cast(str, data).encode("utf-8"))
+        data = buf.getvalue()
 
     res = (session or _get_session()).post(
         url, data=data, headers=headers, timeout=timeout