diff --git a/posthog/__init__.py b/posthog/__init__.py index 5c3335e1..71e4493b 100644 --- a/posthog/__init__.py +++ b/posthog/__init__.py @@ -313,8 +313,11 @@ def get_tags() -> Dict[str, Any]: disabled no-op global client. host: PostHog ingestion host. Defaults to the US ingestion endpoint when not set. - on_error: Optional callback invoked by background consumers when event upload - fails. Keep it short and non-blocking. Lifecycle methods can be called + on_error: Optional callback ``(error, batch)`` invoked when event upload + fails: by background consumers, or on the calling thread in ``sync_mode``. + Capture failures arrive as ``CaptureError``. Without it, each failed batch + logs one aggregate line. A capture that fails inside the callback logs that + line instead of calling it again. Keep it short and non-blocking. Lifecycle methods can be called directly and will be deferred, but the callback must not wait for another thread or task that calls ``flush()``, ``join()``, or ``shutdown()``. debug: Enable verbose SDK logging and re-raise errors from public APIs. diff --git a/posthog/_async_consumer.py b/posthog/_async_consumer.py index 9352139e..0ae74b99 100644 --- a/posthog/_async_consumer.py +++ b/posthog/_async_consumer.py @@ -11,6 +11,7 @@ from ._async_request import async_send_v1_batch from .capture_compression import CaptureCompression +from .capture_send import _CAPTURE_V1_PATH, _capture_loss_message from .consumer import BATCH_SIZE_LIMIT, MAX_MSG_SIZE from .request import DatetimeSerializer @@ -47,6 +48,33 @@ async def _invoke_callback(callback, *args): return result +# True while an SDK-invoked `on_error` runs, so a capture that fails inside the +# callback logs instead of re-entering it. +_IN_ON_ERROR: contextvars.ContextVar[bool] = contextvars.ContextVar( + "posthog_in_on_error", default=False +) + + +async def _report_capture_failure( + on_error: Optional[Callable[..., Any]], + log: logging.Logger, + error: Exception, + batch: list[dict[str, Any]], + endpoint: str, +) -> None: + """Hand a failed send to `on_error`, or log one aggregate line without one.""" + if on_error is None or _IN_ON_ERROR.get(): + log.error(_capture_loss_message(error, max(1, len(batch)), endpoint)) + return + token = _IN_ON_ERROR.set(True) + try: + await _invoke_callback(on_error, error, batch) + except Exception as callback_error: + log.error("on_error handler failed (%s)", type(callback_error).__name__) + finally: + _IN_ON_ERROR.reset(token) + + async def _serialized_event_size(event: dict[str, Any]) -> int: serialized = await asyncio.to_thread(json.dumps, event, cls=DatetimeSerializer) return len(serialized.encode()) @@ -142,18 +170,9 @@ async def upload(self, batch: list[dict[str, Any]]) -> None: try: await self.request(batch) except Exception as error: - self.log.error( - "async capture upload failed (%s, status=%s)", - type(error).__name__, - getattr(error, "status", None), + await _report_capture_failure( + self.on_error, self.log, error, batch, _CAPTURE_V1_PATH ) - if self.on_error: - try: - await _invoke_callback(self.on_error, error, batch) - except Exception as callback_error: - self.log.error( - "on_error handler failed (%s)", type(callback_error).__name__ - ) finally: for _ in batch: self.queue.task_done() diff --git a/posthog/async_client.py b/posthog/async_client.py index e46f0e49..37dfd229 100644 --- a/posthog/async_client.py +++ b/posthog/async_client.py @@ -11,7 +11,7 @@ import weakref from datetime import datetime, timezone from typing import Any, Dict, Mapping, Optional, Union -from uuid import UUID, uuid4 +from uuid import UUID from typing_extensions import Unpack @@ -19,6 +19,7 @@ _STOP, _AsyncConsumer, _invoke_callback, + _report_capture_failure, _is_processing_event, _QueuedEvent, _run_outside_processing_event, @@ -34,6 +35,7 @@ CaptureCompression, _resolve_capture_compression, ) +from .capture_send import _CAPTURE_V1_PATH from .client import ( MAX_DICT_SIZE as _MAX_DICT_SIZE, _MINIMAL_FLAG_CALLED_EVENT_PROPERTIES, @@ -76,7 +78,13 @@ from .release_id import _resolve_release_id from .request import QuotaLimitError, determine_server_host, normalize_host from .types import FlagMetadata, FlagValue, normalize_flags_response -from .utils import SizeLimitedDict, _normalize_timestamp, clean, system_context +from .utils import ( + SizeLimitedDict, + _normalize_timestamp, + _uuid7, + clean, + system_context, +) from .version import VERSION __all__ = ["AsyncClient", "AsyncPosthog"] @@ -380,7 +388,7 @@ def _normalize_uuid(self, msg: dict[str, Any]) -> str: msg["uuid"] = normalized return normalized - normalized = str(uuid4()) + normalized = str(_uuid7()) msg["uuid"] = normalized return normalized @@ -576,20 +584,11 @@ async def capture_immediate( await consumer.request(error_batch) return sent_uuid except Exception as error: - if self.on_error: - try: - await _invoke_callback(self.on_error, error, error_batch) - except Exception as callback_error: - self.log.error( - "on_error handler failed (%s)", type(callback_error).__name__ - ) + await _report_capture_failure( + self.on_error, self.log, error, error_batch, _CAPTURE_V1_PATH + ) if self.debug: raise - self.log.error( - "Immediate async capture failed (%s, status=%s)", - type(error).__name__, - getattr(error, "status", None), - ) return None finally: remaining_calls = self._immediate_callers[current] - 1 diff --git a/posthog/capture_send.py b/posthog/capture_send.py index 862eae09..f254f165 100644 --- a/posthog/capture_send.py +++ b/posthog/capture_send.py @@ -29,7 +29,7 @@ from gzip import GzipFile from io import BytesIO from typing import TYPE_CHECKING, Optional -from uuid import uuid4 +from uuid import UUID from posthog.capture_compression import CaptureCompression, _zstandard from posthog.capture_event import _build_v1_batch_body, _to_v1_event @@ -40,17 +40,17 @@ _get_session, normalize_host, ) -from posthog.utils import remove_trailing_slash +from posthog.utils import _uuid7, remove_trailing_slash if TYPE_CHECKING: import requests log = logging.getLogger("posthog") -# Only the error type is public API: it reaches user code through `on_error` -# callbacks, so callers may want to catch/inspect it. Everything else is +# Only the error types are public API: they reach user code through `on_error` +# callbacks, so callers may want to catch/inspect them. Everything else is # submitter plumbing. -__all__ = ["CaptureError"] +__all__ = ["CaptureError", "CaptureEventResult"] _CAPTURE_V1_PATH = "/i/v1/analytics/events" _CAPTURE_AI_V1_PATH = "/i/v1/ai/events" @@ -80,10 +80,19 @@ # unifies the default with posthog-go/posthog-rs (all 30s). _MAX_BACKOFF_SECONDS = 30 +# First retry delay; it doubles per attempt up to `_MAX_BACKOFF_SECONDS`. Matches +# posthog-go. +_BACKOFF_BASE_SECONDS = 0.1 -@dataclass -class _V1EventResult: - """A single event's directive from a 2xx ``results`` map.""" + +@dataclass(frozen=True) +class CaptureEventResult: + """One event's verdict from a 2xx capture response. + + ``result`` is ``ok``, ``warning``, ``drop`` or ``retry``, or a value newer + than this SDK, which counts as success. ``details`` is the server's reason + tag, such as ``billing_limit_exceeded``. + """ result: Optional[str] details: Optional[str] = None @@ -102,17 +111,22 @@ class _V1ParsedResponse: status_code: int is_success: bool retry_after: Optional[float] = None - results: Optional[dict[str, _V1EventResult]] = None + results: Optional[dict[str, CaptureEventResult]] = None malformed: bool = False error_message: str = "" class CaptureError(APIError): - """Batch-level failure of a capture-v1 send. + """A capture batch that was not fully delivered. - Subclasses :class:`APIError` so the consumer's existing ``on_error`` handling - (which already inspects ``status``/``retry_after``) keeps working; the extra - fields carry v1 specifics for richer logging/callbacks. + Passed to ``on_error`` with the batch. ``status`` is the HTTP status of the + last attempt, or ``0`` when no response arrived (``__cause__`` holds the + transport error). A 2xx ``status`` means the request succeeded but some + events were dropped or ran out of retries. + + ``event_results`` maps each event uuid to its last verdict from a 2xx + response. Events missing from it never got a verdict, because every + attempt that reached them failed as a whole request. """ def __init__( @@ -120,25 +134,101 @@ def __init__( status: int | str, message: str, *, + endpoint: str, retry_after: Optional[float] = None, request_id: Optional[str] = None, attempts: Optional[int] = None, retry_exhausted: Optional[list[str]] = None, drops: Optional[list[tuple[str, Optional[str]]]] = None, + event_results: Optional[dict[str, CaptureEventResult]] = None, ): super().__init__(status, message, retry_after=retry_after) + # Capture path the batch was sent to, so one on_error handler can tell + # the analytics and AI lanes apart. + self.endpoint = endpoint self.request_id = request_id self.attempts = attempts # uuids the server told us to retry but we never delivered (exhausted). self.retry_exhausted = retry_exhausted or [] # (uuid, details) pairs the server told us to drop on a 2xx response. self.drops = drops or [] + self.event_results = event_results or {} + + def verdict_summary(self) -> str: + """Count the undelivered events by verdict and reason, for example + ``drop/billing_limit_exceeded=2, retry/not_persisted=1``. + + Counts the events in ``drops`` and ``retry_exhausted``: events the + server dropped, and events still pending retry after the last 2xx + response. A retry verdict followed by a failed request, for example a + 503 on the last attempt, is not counted; ``event_results`` keeps it. + 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}" + counts[tag] = counts.get(tag, 0) + 1 + return ", ".join(f"{tag}={n}" for tag, n in sorted(counts.items())) def _is_success_status(status: int) -> bool: return 200 <= status < 300 +def _canonical_uuid(value: str) -> str: + """Return ``value`` in the lowercase hyphenated form capture keys results by. + + ``before_send`` can set any uuid form capture parses. A value that does not + parse is returned unchanged; capture rejects its whole batch anyway. + """ + try: + return str(UUID(value)) + except (TypeError, ValueError, AttributeError): + return value + + +def _undelivered_count(error: Exception, batch_size: int) -> int: + if not isinstance(error, CaptureError): + return batch_size + delivered = sum( + 1 + for r in error.event_results.values() + if r.result not in (_RESULT_DROP, _RESULT_RETRY) + ) + return max(0, batch_size - delivered) + + +def _capture_loss_message(error: Exception, batch_size: int, endpoint: str) -> str: + """One aggregate line for a failed batch, for callers with no ``on_error``. + + It never names individual events, includes payloads, or repeats the server's + error text: any of them may carry sensitive content, and per-event lines + scale with event volume. + """ + count = _undelivered_count(error, batch_size) + if isinstance(error, CaptureError): + endpoint = error.endpoint + if isinstance(error.status, int) and _is_success_status(error.status): + detail = f"{len(error.drops)} dropped, {len(error.retry_exhausted)} out of retries" + summary = error.verdict_summary() + if summary: + detail = f"{detail} ({summary})" + return f"{count} event(s) not persisted by {endpoint}: {detail}" + detail = type(error).__name__ + if error.__cause__ is not None: + detail = f"{detail} from {type(error.__cause__).__name__}" + status = getattr(error, "status", None) + if status is not None: + detail = f"{detail} (status={status})" + return f"{count} event(s) not persisted by {endpoint}: {detail}" + + def _parse_retry_after(header_value: Optional[str]) -> Optional[float]: """Parse a ``Retry-After`` header (delta-seconds or HTTP-date) to seconds.""" if not header_value: @@ -239,7 +329,7 @@ def _parse_v1_response(res: "requests.Response") -> _V1ParsedResponse: payload = res.json() raw_results = payload["results"] results = { - uid: _V1EventResult( + uid: CaptureEventResult( result=(r or {}).get("result"), details=(r or {}).get("details"), ) @@ -279,7 +369,7 @@ def _backoff(attempt_index: int, retry_after: Optional[float]) -> None: :data:`_MAX_BACKOFF_SECONDS`, so both sides share one ceiling and a hostile/buggy header can't park the consumer thread. """ - configured = min(2**attempt_index, _MAX_BACKOFF_SECONDS) + configured = min(_BACKOFF_BASE_SECONDS * 2**attempt_index, _MAX_BACKOFF_SECONDS) clamped_retry_after = ( min(retry_after, _MAX_BACKOFF_SECONDS) if retry_after and retry_after > 0 else 0 ) @@ -287,7 +377,7 @@ def _backoff(attempt_index: int, retry_after: Optional[float]) -> None: def _log_result_summary( - request_id: str, attempt: int, results: dict[str, _V1EventResult] + request_id: str, attempt: int, results: dict[str, CaptureEventResult] ) -> None: tally = {_RESULT_OK: 0, _RESULT_WARNING: 0, _RESULT_DROP: 0, _RESULT_RETRY: 0} other = 0 @@ -324,9 +414,11 @@ def _send_v1_batch( ) -> None: """Deliver ``batch`` to the v1 endpoint at ``path`` with partial retry. - 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. + Loops up to ``max_retries + 1`` attempts. After each 2xx it resends only + the events the server tagged ``retry``. ``ok``/``warning`` events, events + with an unrecognized verdict, and events absent from ``results`` succeed + (matching posthog-go and posthog-rs). Results are matched by canonical + uuid, because capture keys them that way whatever form was sent. A server-chosen ``drop`` is a terminal per-event rejection. Drops are accumulated across attempts and surfaced via :class:`CaptureError` even @@ -334,26 +426,47 @@ def _send_v1_batch( 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, - batch-level terminal failure, or retry exhaustion — carrying the accumulated - ``drops`` and any exhausted uuids — so the caller's ``on_error`` fires - unchanged. A transport failure re-raises the underlying exception (drops - collected on an earlier attempt are still tallied in the DEBUG summary). + 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, + so the caller's ``on_error`` gets the full picture. ``request_id`` and the batch ``created_at`` are stable across attempts; ``PostHog-Attempt`` increments. Negative ``max_retries`` values are treated as zero, so delivery is always attempted at least once. """ max_retries = max(0, max_retries) - request_id = str(uuid4()) + request_id = str(_uuid7()) # 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 + pending_uuids = [_canonical_uuid(e["uuid"]) for e in pending_events] # (uuid, details) for every event the server dropped, across all attempts. # Accumulated (not per-attempt) so a drop seen early is not lost when a # later attempt succeeds or clears the outstanding retries. all_drops: list[tuple[str, Optional[str]]] = [] + # Latest 2xx verdict per uuid, across all attempts. + event_results: dict[str, CaptureEventResult] = {} + + def capture_error( + status: int, + message: str, + *, + attempts: int, + retry_after: Optional[float] = None, + retry_exhausted: Optional[list[str]] = None, + ) -> CaptureError: + return CaptureError( + status, + message, + endpoint=path, + retry_after=retry_after, + request_id=request_id, + attempts=attempts, + retry_exhausted=retry_exhausted, + drops=all_drops, + event_results=event_results, + ) for attempt_index in range(max_retries + 1): attempt = attempt_index + 1 @@ -376,10 +489,11 @@ def _send_v1_batch( path=path, ) except Exception as e: - # Transport-level failure (connection/timeout): retry like v0 does. - last_exc = e + # Transport-level failure (connection/timeout): retryable. if last_attempt: - raise + raise capture_error( + 0, f"{type(e).__name__}: {e}", attempts=attempt + ) from e _backoff(attempt_index, None) continue @@ -387,12 +501,10 @@ def _send_v1_batch( if parsed.is_success: if parsed.malformed: - raise CaptureError( + raise capture_error( parsed.status_code, "capture v1 returned a success status with an unparseable body", - request_id=request_id, attempts=attempt, - drops=all_drops, ) results = parsed.results or {} _log_result_summary(request_id, attempt, results) @@ -402,8 +514,11 @@ def _send_v1_batch( 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-rs). + # 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 + event_results[uid] = directive if directive.result == _RESULT_RETRY: retry_events.append(event) retry_uuids.append(uid) @@ -418,22 +533,18 @@ def _send_v1_batch( # surface them via on_error even though the request was a 2xx — # a success status does not mean every event was delivered. if all_drops: - raise CaptureError( + raise capture_error( parsed.status_code, f"{len(all_drops)} event(s) dropped by the server", - request_id=request_id, attempts=attempt, - drops=all_drops, ) return if last_attempt: - raise CaptureError( + raise capture_error( parsed.status_code, f"{len(retry_uuids)} event(s) still pending retry after {attempt} attempt(s)", - request_id=request_id, attempts=attempt, retry_exhausted=retry_uuids, - drops=all_drops, ) pending_events, pending_uuids = retry_events, retry_uuids _backoff(attempt_index, parsed.retry_after) @@ -442,22 +553,13 @@ def _send_v1_batch( # Non-2xx. Retryable transient statuses back off; everything else # (400/401/402/413/415/429/...) is terminal. Any drops collected from a # prior 2xx attempt ride along so on_error still sees them. - v1_error = CaptureError( + v1_error = capture_error( parsed.status_code, parsed.error_message, retry_after=parsed.retry_after, - request_id=request_id, attempts=attempt, - drops=all_drops, ) - if parsed.status_code in _RETRYABLE_STATUSES: - last_exc = v1_error - if last_attempt: - raise v1_error + if parsed.status_code in _RETRYABLE_STATUSES and not last_attempt: _backoff(attempt_index, parsed.retry_after) continue raise v1_error - - # Unreachable in practice (every branch returns or continues). - if last_exc: - raise last_exc diff --git a/posthog/client.py b/posthog/client.py index 611fda2f..a5e23fe2 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -34,6 +34,7 @@ from posthog.capture_send import ( _CAPTURE_AI_V1_PATH, _CAPTURE_V1_PATH, + _capture_loss_message, _send_v1_batch, ) from posthog.consumer import AI_MAX_MSG_SIZE, MAX_MSG_SIZE, Consumer, _DrainSignal @@ -124,6 +125,7 @@ SizeLimitedDict, clean, _normalize_timestamp, + _uuid7, guess_timezone as guess_timezone, system_context, ) @@ -141,6 +143,9 @@ # overrun it by up to the client's request timeout. _TRACES_SHUTDOWN_FLUSH_SECONDS = 30.0 _atexit_deadline: Optional[float] = None +# Marks a thread that is running a sync-mode `on_error`, so a capture that fails +# inside the callback logs instead of re-entering it. +_on_error_state = threading.local() _atexit_deadline_lock = threading.Lock() @@ -246,14 +251,13 @@ def _stringify_event_uuid(value) -> str: ) try: - UUID(stringified) + # Canonical form, because capture keys per-event results by it. + return str(UUID(stringified)) except ValueError: raise ValueError( f"Invalid event uuid {value!r}. Expected a valid UUID string or uuid.UUID instance." ) from None - return stringified - def add_context_tags(properties): properties = properties or {} @@ -724,8 +728,12 @@ def __init__( API methods. max_queue_size: Maximum number of events buffered before upload. send: If False, queueing succeeds but events are not sent. - on_error: Optional callback invoked by background consumers when an - upload fails. Keep it short and non-blocking. Calling lifecycle + on_error: Optional callback ``(error, batch)`` invoked when an upload + fails: by background consumers, or on the calling thread in + ``sync_mode``. Capture failures arrive as ``CaptureError``. + Without it, each failed batch logs one aggregate line. A capture + that fails inside the callback logs that line instead of calling + it again. Keep it short and non-blocking. Calling lifecycle methods directly is safe and deferred, but do not start another thread or task that calls ``flush()``, ``join()``, or ``shutdown()`` and then wait for it from the callback. @@ -2310,7 +2318,26 @@ def _normalize_event_uuid(self, msg): if "uuid" not in msg: # Always send a uuid, so we can always return one - msg["uuid"] = stringify_id(uuid4()) + msg["uuid"] = str(_uuid7()) + + def _report_capture_failure( + self, error: Exception, batch: list[dict], endpoint: str + ) -> None: + """Hand a failed sync-mode send to `on_error`, like the queued path does. + + Without a handler, or when the failed capture ran inside `on_error`, + log one aggregate line instead, so the loss stays visible without recursion. + """ + if not self.on_error or getattr(_on_error_state, "active", False): + self.log.error(_capture_loss_message(error, len(batch), endpoint)) + return + _on_error_state.active = True + try: + self.on_error(error, batch) + except Exception as callback_error: + self.log.error("on_error handler failed: %s", callback_error) + finally: + _on_error_state.active = False def _enqueue(self, msg, disable_geoip, lane=None, property_allowlist=None): # type: (...) -> Optional[str] @@ -2423,7 +2450,14 @@ def send_sync() -> None: path=lane.endpoint, ) - if lane.run_sync_if_open(send_sync): + try: + admitted = lane.run_sync_if_open(send_sync) + except Exception as e: + self._report_capture_failure(e, [msg], lane.endpoint) + if self.debug: + raise + return None + if admitted: return sent_uuid self.log.warning( "%s lane received event %s after shutdown, dropping it", diff --git a/posthog/consumer.py b/posthog/consumer.py index daad58c3..22944064 100644 --- a/posthog/consumer.py +++ b/posthog/consumer.py @@ -6,7 +6,11 @@ from posthog._logging import _configure_posthog_logging from posthog.capture_compression import CaptureCompression -from posthog.capture_send import _CAPTURE_V1_PATH, _send_v1_batch +from posthog.capture_send import ( + _CAPTURE_V1_PATH, + _capture_loss_message, + _send_v1_batch, +) from posthog.request import ( USER_AGENT as _USER_AGENT, DatetimeSerializer, @@ -105,7 +109,7 @@ def __init__( host=None, on_error=None, flush_interval=5.0, - retries=10, + retries=3, timeout=15, historical_migration=False, endpoint=_CAPTURE_V1_PATH, @@ -172,13 +176,14 @@ def upload(self): self.request(batch) success = True except Exception as e: - self.log.error("error uploading: %s", e) success = False if self.on_error: try: self.on_error(e, batch) - except Exception as e: - self.log.error("on_error handler failed: %s", e) + except Exception as callback_error: + self.log.error("on_error handler failed: %s", callback_error) + else: + self.log.error(_capture_loss_message(e, len(batch), self.endpoint)) finally: # mark items as acknowledged from queue for item in batch: diff --git a/posthog/test/test_async_client.py b/posthog/test/test_async_client.py index 34c098d7..02ce2ce4 100644 --- a/posthog/test/test_async_client.py +++ b/posthog/test/test_async_client.py @@ -7,6 +7,7 @@ import sys import threading from unittest import mock +from uuid import UUID import pytest @@ -56,7 +57,7 @@ async def send_batch(api_key, host, batch, **kwargs): distinct_id="user-1", properties={"plan": "pro"}, ) - assert isinstance(event_uuid, str) + assert UUID(event_uuid).version == 7 await client.flush(timeout_seconds=1) assert len(batches) == 1 @@ -706,6 +707,34 @@ def on_error(error, batch): assert callback_thread != threading.get_ident() +@pytest.mark.asyncio +@pytest.mark.parametrize("immediate", [False, True]) +async def test_capture_failure_inside_on_error_logs_instead_of_recursing( + caplog, immediate +): + calls = [] + client: AsyncPosthog + + async def on_error(error, batch): + calls.append(batch[0]["event"]) + await client.capture_immediate("from_callback", distinct_id="user-1") + + with patch_async_capture_send(side_effect=APIError(503, "unavailable")): + client = AsyncPosthog("test-key", on_error=on_error, flush_at=1, max_retries=0) + if immediate: + await client.capture_immediate("event", distinct_id="user-1") + else: + client.capture("event", distinct_id="user-1") + await client.flush(timeout_seconds=1) + await client.shutdown() + + assert calls == ["event"] + assert ( + "1 event(s) not persisted by /i/v1/analytics/events: APIError (status=503)" + in caplog.text + ) + + @pytest.mark.asyncio @pytest.mark.parametrize("immediate", [False, True]) async def test_failed_capture_does_not_log_server_response_detail(caplog, immediate): diff --git a/posthog/test/test_capture_send.py b/posthog/test/test_capture_send.py index 637eee85..1f8c9506 100644 --- a/posthog/test/test_capture_send.py +++ b/posthog/test/test_capture_send.py @@ -3,6 +3,7 @@ import zlib from datetime import datetime, timedelta from unittest import mock +from uuid import UUID import zstandard @@ -19,6 +20,8 @@ _HEADER_SDK_INFO, _MAX_BACKOFF_SECONDS, CaptureError, + CaptureEventResult, + _capture_loss_message, _parse_v1_response, _post_v1, _send_v1_batch, @@ -317,6 +320,7 @@ def test_request_id_and_created_at_stable_attempt_increments(self) -> None: ], ) self.assertEqual(stub.calls[0]["request_id"], stub.calls[1]["request_id"]) + self.assertEqual(UUID(stub.calls[0]["request_id"]).version, 7) # created_at is hoisted once, so the envelope timestamp is identical # across retry attempts (only the attempt header increments). self.assertEqual(stub.calls[0]["created_at"], stub.calls[1]["created_at"]) @@ -383,18 +387,56 @@ def test_retry_exhausted_raises_with_uuids(self) -> None: def test_retry_exhausted_carries_earlier_drops(self) -> None: # A drop seen on attempt 1 rides along on the retry-exhaustion error. - batch = [_msg("u-drop"), _msg("u-retry")] + batch = [_msg("u-ok"), _msg("u-drop"), _msg("u-retry")] stub, exc = self._run_expecting_error( batch, [ - _results_response({"u-drop": ("drop", "billing"), "u-retry": "retry"}), - _results_response({"u-retry": "retry"}), + _results_response( + { + "u-ok": "ok", + "u-drop": ("drop", "billing"), + "u-retry": "retry", + } + ), + _results_response({"u-retry": ("retry", "not_persisted")}), ], max_retries=1, ) self.assertEqual(len(stub.calls), 2) + self.assertEqual(exc.endpoint, _CAPTURE_V1_PATH) self.assertEqual(exc.retry_exhausted, ["u-retry"]) self.assertEqual(exc.drops, [("u-drop", "billing")]) + self.assertEqual( + exc.event_results, + { + "u-ok": CaptureEventResult("ok"), + "u-drop": CaptureEventResult("drop", "billing"), + "u-retry": CaptureEventResult("retry", "not_persisted"), + }, + ) + self.assertEqual(exc.verdict_summary(), "drop/billing=1, retry/not_persisted=1") + + @parameterized.expand( + [ + ("uppercase", "AAAAAAAA-AAAA-4AAA-8AAA-AAAAAAAAAAAA"), + ("no_hyphens", "aaaaaaaaaaaa4aaa8aaaaaaaaaaaaaaa"), + ("braced", "{aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa}"), + ("urn", "urn:uuid:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"), + ] + ) + def test_results_match_non_canonical_uuid(self, _name, sent_uuid) -> None: + # Capture parses any of these forms but keys results canonically, so a + # verdict for a uuid `before_send` rewrote must still reach its event. + canonical = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" + stub, exc = self._run_expecting_error( + [_msg(sent_uuid)], + [ + _results_response({canonical: "retry"}), + _results_response({canonical: ("drop", "billing")}), + ], + ) + self.assertEqual(len(stub.calls), 2) + self.assertEqual(exc.drops, [(canonical, "billing")]) def test_malformed_2xx_is_terminal(self) -> None: stub, exc = self._run_expecting_error( @@ -438,14 +480,22 @@ def test_transport_error_then_success(self) -> None: ) self.assertEqual(len(stub.calls), 2) - def test_transport_error_exhausted_reraises_original(self) -> None: + def test_transport_error_exhausted_raises_capture_error(self) -> None: stub = _PostV1Stub([ConnectionError("boom"), ConnectionError("boom")]) with mock.patch("posthog.capture_send._post_v1", stub): - with self.assertRaises(ConnectionError): + with self.assertRaises(CaptureError) as ctx: _send_v1_batch( - "phc_key", "https://app.posthog.com", [_msg("u-1")], max_retries=1 + "phc_key", + "https://app.posthog.com", + [_msg("u-1")], + max_retries=1, + path=_CAPTURE_AI_V1_PATH, ) self.assertEqual(len(stub.calls), 2) + self.assertEqual(ctx.exception.status, 0) + self.assertEqual(ctx.exception.endpoint, _CAPTURE_AI_V1_PATH) + self.assertEqual(ctx.exception.attempts, 2) + self.assertIsInstance(ctx.exception.__cause__, ConnectionError) def test_negative_max_retries_still_attempts_delivery_once(self) -> None: stub = _PostV1Stub([_results_response({"u-1": "ok"})]) @@ -461,20 +511,69 @@ def test_negative_max_retries_still_attempts_delivery_once(self) -> None: def test_small_retry_after_does_not_shorten_backoff(self) -> None: # A Retry-After smaller than the configured backoff must not make the # client retry earlier than its own schedule (Retry-After is a minimum). - # attempt_index=1 -> configured backoff 2s; Retry-After 0.5s is ignored. + # attempt_index=1 -> configured backoff 0.2s; Retry-After 0.1s is ignored. stub = self._run( [_msg("u-1")], [ _results_response({"u-1": "retry"}), - _results_response({"u-1": "retry"}, headers={"Retry-After": "0.5"}), + _results_response({"u-1": "retry"}, headers={"Retry-After": "0.1"}), _results_response({"u-1": "ok"}), ], max_retries=3, ) self.assertEqual(len(stub.calls), 3) - # First backoff (attempt_index 0) waits 1s; second (attempt_index 1) - # keeps the 2s configured backoff rather than the smaller 0.5s header. - self.assertEqual([c.args[0] for c in self.sleep.call_args_list], [1, 2]) + # First backoff (attempt_index 0) waits 0.1s; second (attempt_index 1) + # keeps the 0.2s configured backoff rather than the smaller 0.1s header. + self.assertEqual([c.args[0] for c in self.sleep.call_args_list], [0.1, 0.2]) + + +class TestCaptureLossMessage(unittest.TestCase): + @parameterized.expand( + [ + ( + "partial_2xx", + CaptureError( + 200, + "2 event(s) not delivered", + endpoint=_CAPTURE_V1_PATH, + drops=[("u-drop", "billing")], + retry_exhausted=["u-retry"], + event_results={ + "u-ok": CaptureEventResult("ok"), + "u-drop": CaptureEventResult("drop", "billing"), + "u-retry": CaptureEventResult("retry", "not_persisted"), + }, + ), + 3, + "2 event(s) not persisted by /i/v1/analytics/events: 1 dropped, " + "1 out of retries (drop/billing=1, retry/not_persisted=1)", + ), + ( + "request_failure_after_partial_success", + CaptureError( + 503, + "unavailable", + endpoint=_CAPTURE_AI_V1_PATH, + event_results={ + "u-ok": CaptureEventResult("warning"), + "u-retry": CaptureEventResult("retry"), + }, + ), + 2, + "1 event(s) not persisted by /i/v1/ai/events: CaptureError (status=503)", + ), + ( + "other_exception_uses_caller_endpoint", + ValueError("bad payload"), + 4, + "4 event(s) not persisted by /i/v1/analytics/events: ValueError", + ), + ] + ) + def test_message(self, _name, error, batch_size, expected) -> None: + self.assertEqual( + _capture_loss_message(error, batch_size, _CAPTURE_V1_PATH), expected + ) class TestBackoff(unittest.TestCase): @@ -483,13 +582,13 @@ class TestBackoff(unittest.TestCase): @parameterized.expand( [ # (attempt_index, retry_after, expected sleep seconds) - ("first_no_header", 0, None, 1), - ("second_no_header", 1, None, 2), + ("first_no_header", 0, None, 0.1), + ("second_no_header", 1, None, 0.2), ("exp_capped_at_30", 10, None, 30), - ("zero_header_uses_backoff", 0, 0, 1), + ("zero_header_uses_backoff", 0, 0, 0.1), ("larger_header_wins", 0, 5.0, 5.0), - ("smaller_header_ignored", 3, 2.0, 8), # configured 8 > 2.0 - ("equal_header_and_backoff", 0, 1.0, 1), + ("smaller_header_ignored", 3, 0.5, 0.8), # configured 0.8 > 0.5 + ("equal_header_and_backoff", 1, 0.2, 0.2), ("header_at_ceiling", 0, 30.0, 30), ("header_above_ceiling_clamped", 0, 120.0, _MAX_BACKOFF_SECONDS), ("absurd_header_clamped", 0, 10**9, _MAX_BACKOFF_SECONDS), @@ -498,4 +597,5 @@ class TestBackoff(unittest.TestCase): def test_backoff(self, _name, attempt_index, retry_after, expected) -> None: with mock.patch("posthog.capture_send.time.sleep") as sleep: _backoff(attempt_index, retry_after) - sleep.assert_called_once_with(expected) + sleep.assert_called_once() + self.assertAlmostEqual(sleep.call_args.args[0], expected) diff --git a/posthog/test/test_client.py b/posthog/test/test_client.py index 142740ad..37d862b5 100644 --- a/posthog/test/test_client.py +++ b/posthog/test/test_client.py @@ -470,6 +470,24 @@ def test_basic_capture_with_uuid_object(self): msg = sent_batch(mock_post)[0] self.assertEqual(msg["uuid"], str(uuid)) + @parameterized.expand( + [ + ("uppercase", "0190A3B4-C5D6-7E8F-9A0B-1C2D3E4F5A6B"), + ("no hyphens", "0190a3b4c5d67e8f9a0b1c2d3e4f5a6b"), + ("braced", "{0190a3b4-c5d6-7e8f-9a0b-1c2d3e4f5a6b}"), + ] + ) + def test_capture_sends_and_returns_canonical_uuid(self, _name, supplied): + with patch_capture_send("client") as mock_post: + client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, sync_mode=True) + msg_uuid = client.capture( + "python test event", distinct_id="distinct_id", uuid=supplied + ) + + canonical = "0190a3b4-c5d6-7e8f-9a0b-1c2d3e4f5a6b" + self.assertEqual(msg_uuid, canonical) + self.assertEqual(sent_batch(mock_post)[0]["uuid"], canonical) + @parameterized.expand( [ ("empty string", ""), @@ -489,7 +507,7 @@ def test_capture_with_invalid_uuid_logs_and_falls_back_to_generated_uuid( ) self.assertIsNotNone(msg_uuid) - UUID(msg_uuid) + self.assertEqual(UUID(msg_uuid).version, 7) mock_post.assert_called_once() msg = sent_batch(mock_post)[0] self.assertEqual(msg["uuid"], msg_uuid) @@ -2707,20 +2725,79 @@ def flush(timeout_seconds): self.assertEqual(timeouts, [0, None]) - def test_sync_send_failure_does_not_invoke_async_on_error(self): - on_error = mock.Mock() + def test_sync_send_failure_invokes_on_error_after_releasing_send_slot(self): + error = Exception("upload failed") + active_sends_in_callback = [] + client: Client + + def on_error(err, batch): + active_sends_in_callback.append(client._analytics_lane._active_sync_sends) + self.assertIs(err, error) + self.assertEqual([event["event"] for event in batch], ["event"]) + client.shutdown() + + client = Client(FAKE_TEST_API_KEY, sync_mode=True, on_error=on_error) + + with patch_capture_send("client", side_effect=error): + result = client.capture("event", distinct_id="distinct_id") + + self.assertIsNone(result) + self.assertEqual(active_sends_in_callback, [0]) + + def test_sync_send_failure_inside_on_error_logs_instead_of_recursing(self): + calls = [] + client: Client + + def on_error(err, batch): + calls.append(batch[0]["event"]) + client.capture("from_callback", distinct_id="distinct_id") + + client = Client(FAKE_TEST_API_KEY, sync_mode=True, on_error=on_error) + + with patch_capture_send("client", side_effect=Exception("upload failed")): + with self.assertLogs("posthog", level="ERROR") as logs: + client.capture("event", distinct_id="distinct_id") + + self.assertEqual(calls, ["event"]) + self.assertIn( + "1 event(s) not persisted by /i/v1/analytics/events: Exception", + "\n".join(logs.output), + ) + + @parameterized.expand([("capture",), ("capture_ai",)]) + def test_sync_send_failure_in_debug_reports_once_then_raises(self, method_name): + errors = [] client = Client( FAKE_TEST_API_KEY, sync_mode=True, - on_error=on_error, + debug=True, + on_error=lambda err, batch: errors.append(err), ) + failure = APIError(503, "unavailable") - with patch_capture_send("client", side_effect=Exception("upload failed")): - result = client.capture("event", distinct_id="distinct_id") + with patch_capture_send("client", side_effect=failure): + with self.assertRaises(APIError) as raised: + getattr(client, method_name)("event", distinct_id="distinct_id") + + self.assertIs(raised.exception, failure) + self.assertEqual(errors, [failure]) + + def test_sync_send_failure_without_on_error_logs_aggregate_line(self): + client = Client(FAKE_TEST_API_KEY, sync_mode=True) + + with patch_capture_send( + "client", side_effect=APIError(400, "password=server-secret") + ): + with self.assertLogs("posthog", level="ERROR") as logs: + result = client.capture("event", distinct_id="distinct_id") self.assertIsNone(result) - on_error.assert_not_called() - self.assertEqual(client._analytics_lane._active_sync_sends, 0) + output = "\n".join(logs.output) + self.assertIn( + "1 event(s) not persisted by /i/v1/analytics/events: APIError (status=400)", + output, + ) + self.assertNotIn("server-secret", output) def test_on_error_can_request_shutdown_with_pending_work(self): first_send_started = threading.Event() diff --git a/posthog/test/test_consumer.py b/posthog/test/test_consumer.py index 2274c690..ca474882 100644 --- a/posthog/test/test_consumer.py +++ b/posthog/test/test_consumer.py @@ -232,9 +232,11 @@ def test_message_only_error_logs_include_posthog_prefix(self) -> None: # other tests can log into the same stream. Assert on the line under # test rather than on the entire capture. upload_logs = [ - line for line in logs.getvalue().splitlines() if "error uploading" in line + line for line in logs.getvalue().splitlines() if "not persisted" in line ] - expected_log = "[PostHog] error uploading: boom" + expected_log = ( + "[PostHog] 1 event(s) not persisted by /i/v1/analytics/events: Exception" + ) self.assertEqual( [line for line in upload_logs if line == expected_log], [expected_log] ) diff --git a/posthog/test/test_utils.py b/posthog/test/test_utils.py index e655c8cd..c2dd1a0c 100644 --- a/posthog/test/test_utils.py +++ b/posthog/test/test_utils.py @@ -261,6 +261,32 @@ def test_clean_fn(self): def test_remove_slash(self, input_url, expected_url): assert expected_url == utils.remove_trailing_slash(input_url) + @parameterized.expand( + [ + ("random_ones", b"\xff" * 10, "01234567-89ab-7fff-bfff-ffffffffffff"), + ("random_zeros", b"\x00" * 10, "01234567-89ab-7000-8000-000000000000"), + ( + "random_mixed", + bytes(range(0x10, 0x1A)), + "01234567-89ab-7011-9213-141516171819", + ), + ] + ) + def test_uuid7_layout(self, _name, random_bytes, expected): + # Bit 48 of the millisecond clock is set to check that it is masked off. + time_ns = ((1 << 48) + 0x0123456789AB) * 1_000_000 + 999_999 + with ( + mock.patch.object(utils.time, "time_ns", return_value=time_ns), + mock.patch.object( + utils.os, + "urandom", + side_effect=lambda n: (random_bytes + b"\xaa" * n)[:n], + ), + ): + value = utils._uuid7() + self.assertEqual(str(value), expected) + self.assertEqual(value.version, 7) + def test_clean_pydantic(self): class ModelV2(BaseModel): foo: str diff --git a/posthog/utils.py b/posthog/utils.py index 05608983..84047841 100644 --- a/posthog/utils.py +++ b/posthog/utils.py @@ -1,6 +1,7 @@ import json import logging import numbers +import os import re import threading import time @@ -51,6 +52,21 @@ def guess_timezone(dt: datetime) -> datetime: return dt.astimezone(timezone.utc) +def _uuid7() -> UUID: + """Return a time-ordered version 7 UUID (RFC 9562). + + Capture generates v7 server-side when a client omits the uuid, so + SDK-generated event uuids sort the same way. + """ + unix_ms = time.time_ns() // 1_000_000 + # Python 3.10 requires the byte order; from 3.11 "big" is the default. + random_bits = int.from_bytes(os.urandom(10), "big") # pragma: no mutate + value = ((unix_ms & ((1 << 48) - 1)) << 80) | random_bits + value = (value & ~(0xF << 76)) | (0x7 << 76) + value = (value & ~(0x3 << 62)) | (0x2 << 62) + return UUID(int=value) + + def _normalize_timestamp(timestamp: Union[datetime, str]) -> str: """Normalize a datetime or parseable ISO datetime string to UTC.""" parsed_timestamp: datetime diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index c6165d97..0cb0bc89 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -692,8 +692,12 @@ attribute posthog.capture_compression.CaptureCompression.ZSTD = 'zstd' attribute posthog.capture_exception_code_variables = False attribute posthog.capture_send.CaptureError.attempts = attempts attribute posthog.capture_send.CaptureError.drops = drops or [] +attribute posthog.capture_send.CaptureError.endpoint = endpoint +attribute posthog.capture_send.CaptureError.event_results = event_results or {} attribute posthog.capture_send.CaptureError.request_id = request_id attribute posthog.capture_send.CaptureError.retry_exhausted = retry_exhausted or [] +attribute posthog.capture_send.CaptureEventResult.details: Optional[str] = None +attribute posthog.capture_send.CaptureEventResult.result: Optional[str] attribute posthog.capture_trace_context = False attribute posthog.client.Client.api_key = (project_api_key or '').strip() attribute posthog.client.Client.capture_compression = _resolve_capture_compression(capture_compression) @@ -1164,9 +1168,10 @@ class posthog.async_client.AsyncClient(project_api_key: str, host: Optional[str] class posthog.async_client.AsyncPosthog class posthog.bucketed_rate_limiter.BucketedRateLimiter(bucket_size: Number, refill_rate: Number, refill_interval_seconds: Number, on_bucket_rate_limited: Optional[Callable[[Hashable], None]] = None, clock: Callable[[], float] = time.monotonic) class posthog.capture_compression.CaptureCompression -class posthog.capture_send.CaptureError(status: int | str, message: str, *, retry_after: Optional[float] = None, request_id: Optional[str] = None, attempts: Optional[int] = None, retry_exhausted: Optional[list[str]] = None, drops: Optional[list[tuple[str, Optional[str]]]] = None) +class posthog.capture_send.CaptureError(status: int | str, message: str, *, endpoint: str, retry_after: Optional[float] = None, request_id: Optional[str] = None, attempts: Optional[int] = None, retry_exhausted: Optional[list[str]] = None, drops: Optional[list[tuple[str, Optional[str]]]] = None, event_results: Optional[dict[str, CaptureEventResult]] = None) +class posthog.capture_send.CaptureEventResult(result: Optional[str], details: Optional[str] = None) class posthog.client.Client(project_api_key: str, host=None, *, debug=False, max_queue_size=10000, send=True, on_error=None, flush_at=100, flush_interval=5.0, max_retries=3, sync_mode=False, timeout=15, thread=1, poll_interval=30, personal_api_key=None, disabled=False, disable_geoip=True, is_server=True, historical_migration=False, feature_flags_request_timeout_seconds=3, feature_flags_request_max_retries=1, super_properties=None, enable_exception_autocapture=False, log_captured_exceptions=False, project_root=None, privacy_mode=False, before_send=None, flag_fallback_cache_url=None, enable_local_evaluation=True, flag_definition_cache_provider: Optional[FlagDefinitionCacheProvider] = None, capture_exception_code_variables=False, code_variables_mask_patterns=None, code_variables_ignore_patterns=None, code_variables_mask_url_credentials=None, code_variables_detect_secrets=None, in_app_modules: list[str] | None = None, enable_exception_autocapture_rate_limiting=False, exception_autocapture_bucket_size=ExceptionCapture.DEFAULT_BUCKET_SIZE, exception_autocapture_refill_rate=ExceptionCapture.DEFAULT_REFILL_RATE, exception_autocapture_refill_interval_seconds=ExceptionCapture.DEFAULT_REFILL_INTERVAL_SECONDS, capture_compression: Optional[Union[CaptureCompression, str]] = None, secret_key=None, metrics: Optional[dict] = None, enable_full_ai_capture=False, capture_trace_context=False, _use_ai_lane=False, _enable_multimodal_capture=False, traces: Optional[dict] = None) -class posthog.consumer.Consumer(queue, api_key, flush_at=100, host=None, on_error=None, flush_interval=5.0, retries=10, timeout=15, historical_migration=False, endpoint=_CAPTURE_V1_PATH, max_msg_size=MAX_MSG_SIZE, capture_compression=CaptureCompression.NONE) +class posthog.consumer.Consumer(queue, api_key, flush_at=100, host=None, on_error=None, flush_interval=5.0, retries=3, timeout=15, historical_migration=False, endpoint=_CAPTURE_V1_PATH, max_msg_size=MAX_MSG_SIZE, capture_compression=CaptureCompression.NONE) class posthog.contexts.ContextScope(parent=None, fresh: bool = False, capture_exceptions: bool = True, client: Optional[Client] = None) class posthog.exception_capture.ExceptionCapture(client: Client, rate_limiting_enabled=False, bucket_size=DEFAULT_BUCKET_SIZE, refill_rate=DEFAULT_REFILL_RATE, refill_interval_seconds=DEFAULT_REFILL_INTERVAL_SECONDS) class posthog.exception_utils.AnnotatedValue(value, metadata) @@ -1559,6 +1564,7 @@ method posthog.async_client.AsyncClient.set_once(**kwargs: Unpack[OptionalSetArg method posthog.async_client.AsyncClient.shutdown() -> None method posthog.bucketed_rate_limiter.BucketedRateLimiter.consume_rate_limit(key: Hashable) -> bool method posthog.bucketed_rate_limiter.BucketedRateLimiter.stop() -> None +method posthog.capture_send.CaptureError.verdict_summary() -> str method posthog.client.Client.alias(previous_id: ID_TYPES, distinct_id: Optional[str], timestamp: Optional[Union[datetime, str]] = None, uuid: Optional[str] = None, disable_geoip: Optional[bool] = None) -> Optional[str] method posthog.client.Client.capture(event: str, **kwargs: Unpack[OptionalCaptureArgs]) -> Optional[str] method posthog.client.Client.capture_ai(event: str, **kwargs: Unpack[OptionalCaptureArgs]) -> Optional[str]