fix: own capture connection retries and redact hook error logs

.sampo/changesets/capture-hook-log-and-connection-retries.md

@@ -0,0 +1,9 @@
+---
+pypi/posthog: patch
+---
+
+Capture makes one connection try per attempt, so every physical request carries its own attempt number and request timestamp. 7.x also retried a failed connection up to twice inside each attempt, reusing the request timestamp, which shifted the events' stored time by the time spent connecting.
+
+When `before_send` raises, the error line names the event and the exception type, not the exception message. The traceback is logged at debug level.
+
+Events without `NaN` or `Infinity` are encoded once, without first copying them to replace non-finite numbers. Events that contain them still send those values as `null`.

docs/migration-7.x-to-8.0.md

@@ -103,8 +103,9 @@ When the whole request is over the billing limit, capture returns a `402` with n
 - In `sync_mode`, a failed `capture()` calls `on_error` and returns `None`. With `debug=True` it calls `on_error` and then re-raises the error.
 - If a capture inside `on_error` fails, the SDK logs the failure and does not call `on_error` again.
 - The SDK retries `408`, `500`, `502`, `503` and `504` responses and network errors. Other error statuses fail the batch at once, including `429` and the other 5xx codes. 7.x also retried `429` and every 5xx. Capture itself does not return `429`; a proxy in front of it can.
+- Each attempt makes one connection try. 7.x also retried a failed connection up to twice inside each attempt, so a capture request could block for up to three times the `timeout` per attempt.
 - Retries start at 100 ms and double each attempt, up to 30 seconds. Each wait varies randomly by up to 25%, so many clients do not retry at the same moment. A `Retry-After` header sets a minimum wait, up to 30 seconds. The `Consumer` default is 3 retries, down from 10.
-- The SDK's own log lines no longer contain event content: the debug line for a queued event names the event and the lane, and the invalid timestamp warning no longer includes the value. An exception your `before_send` hook raises is still logged with its message. To inspect events, use `before_send`.
+- The SDK's own log lines no longer contain event content: the debug line for a queued event names the event and the lane, and the invalid timestamp warning no longer includes the value. When your `before_send` hook raises, the error line names the event and the exception type, not the exception message, which can hold the data the hook was removing. The traceback is logged at debug level. To inspect events, use `before_send`.
 - A personal API key (`phx_...`) passed as the project API key disables the client and logs one error, as an empty key does. Capture rejects personal API keys, and the SDK no longer sends one with every request.
 
 ## Invalid events
@@ -181,6 +182,7 @@ Capture v1 sends processing options in an `options` object, next to `properties`
   | --- | --- |
   | `$cookieless_mode` | `cookieless_mode` |
   | `$ignore_sent_at` | `disable_skew_correction` |
+  | `$product_tour_id` | `product_tour_id` |
   | `$process_person_profile` | `process_person_profile` |
 
 - An option set at any layer wins over its legacy property set at any layer. For example, `super_options={"cookieless_mode": True}` wins over an event's `$cookieless_mode: False`. When you move a default to options, move the per-event overrides of that key to options too.

posthog/async_client.py

@@ -105,6 +105,8 @@
     clean,
     _drop_invalid_distinct_id,
     _drop_invalid_event_name,
+    _log_before_send_bad_return,
+    _log_before_send_error,
     system_context,
 )
 from .version import VERSION
@@ -524,18 +526,20 @@ async def _run_before_send(self, msg: dict[str, Any]) -> Optional[dict[str, Any]
             return msg
 
         original_uuid = msg["uuid"]
+        event_name = msg.get("event")
         try:
             result = await _invoke_callback(self.before_send, msg)
             if result is None:
                 self.log.debug("event dropped by before_send callback")
                 return None
             if not isinstance(result, dict):
-                raise TypeError("before_send must return a dict or None")
+                _log_before_send_bad_return(self.log, event_name, result)
+                return None
             processed = clean(result)
             processed["uuid"] = original_uuid
             return processed
         except Exception as error:
-            self.log.error("Error in before_send callback (%s)", type(error).__name__)
+            _log_before_send_error(self.log, event_name, error)
             return None
 
     def _event_defaults(

posthog/capture_send.py

@@ -44,7 +44,7 @@
     DatetimeSerializer,
     USER_AGENT,
     APIError,
-    _get_session,
+    _get_capture_session,
     normalize_host,
 )
 from posthog.utils import _uuid7, remove_trailing_slash
@@ -376,7 +376,7 @@ def _post_v1(
         headers["Content-Encoding"] = encoding
 
     log.debug("capture v1 POST %s attempt=%s request_id=%s", url, attempt, request_id)
-    http = session or _get_session()
+    http = session or _get_capture_session()
     for _ in range(_MAX_REDIRECTS + 1):
         res = http.post(
             url, data=body, headers=headers, timeout=timeout, allow_redirects=False

posthog/client.py

@@ -142,6 +142,8 @@
     _is_personal_api_key,
     _drop_invalid_distinct_id,
     _drop_invalid_event_name,
+    _log_before_send_bad_return,
+    _log_before_send_error,
     guess_timezone as guess_timezone,
     system_context,
 )
@@ -2601,16 +2603,18 @@ def _enqueue(
         msg = clean(msg)
 
         if self.before_send:
+            event_name = msg.get("event")
             try:
                 modified_msg = self.before_send(msg)
                 if modified_msg is None:
                     self.log.debug("Event dropped by before_send callback")
                     return None
                 if not isinstance(modified_msg, dict):
-                    raise TypeError("before_send must return a dict or None")
+                    _log_before_send_bad_return(self.log, event_name, modified_msg)
+                    return None
                 msg = clean(modified_msg)
             except Exception as e:
-                self.log.exception(f"Error in before_send callback: {e}")
+                _log_before_send_error(self.log, event_name, e)
                 return None
 
         # Re-normalized after before_send, which may have replaced or removed

posthog/request.py

@@ -73,7 +73,7 @@ def init_poolmanager(self, *args, **kwargs):
 
 
 def _build_session(socket_options: Optional[SocketOptions] = None) -> requests.Session:
-    """Build a session for general requests (batch, remote config, etc.)."""
+    """Build a session for general requests (remote config, flag definitions, etc.)."""
     adapter = HTTPAdapterWithSocketOptions(
         max_retries=Retry(
             total=2,
@@ -87,13 +87,13 @@ def _build_session(socket_options: Optional[SocketOptions] = None) -> requests.S
     return session
 
 
-def _build_flags_session(
+def _build_no_retry_session(
     socket_options: Optional[SocketOptions] = None,
 ) -> requests.Session:
-    """Build a session for feature flag requests.
+    """Build a session whose adapter never retries; ``flags()`` and capture own retries.
 
-    /flags retries are handled explicitly in ``flags()`` so that only
-    transport failures and contract-defined transient HTTP responses are retried.
+    Capture needs a fresh ``PostHog-Request-Timestamp`` on every physical attempt,
+    because capture uses it to correct clock skew.
     """
     adapter = HTTPAdapterWithSocketOptions(
         max_retries=Retry(total=0, connect=0, read=0, status=0),
@@ -105,7 +105,8 @@ def _build_flags_session(
 
 
 _session = _build_session()
-_flags_session = _build_flags_session()
+_flags_session = _build_no_retry_session()
+_capture_session = _build_no_retry_session()
 _socket_options: Optional[SocketOptions] = None
 _pooling_enabled = True
 
@@ -119,21 +120,30 @@ def _get_session() -> requests.Session:
 def _get_flags_session() -> requests.Session:
     if _pooling_enabled:
         return _flags_session
-    return _build_flags_session(_socket_options)
+    return _build_no_retry_session(_socket_options)
+
+
+def _get_capture_session() -> requests.Session:
+    if _pooling_enabled:
+        return _capture_session
+    return _build_no_retry_session(_socket_options)
 
 
 def reset_sessions() -> None:
     """
     Reset the global sessions. This should be called after a fork to ensure
     that the child process does not use the parent's connection pool.
     """
-    global _session, _flags_session
+    global _session, _flags_session, _capture_session
     if _session:
         _session.close()
     if _flags_session:
         _flags_session.close()
+    if _capture_session:
+        _capture_session.close()
     _session = _build_session(_socket_options)
-    _flags_session = _build_flags_session(_socket_options)
+    _flags_session = _build_no_retry_session(_socket_options)
+    _capture_session = _build_no_retry_session(_socket_options)
 
 
 def set_socket_options(socket_options: Optional[SocketOptions]) -> None:
@@ -151,12 +161,13 @@ def set_socket_options(socket_options: Optional[SocketOptions]) -> None:
         from posthog import set_socket_options
         set_socket_options([(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)])
     """
-    global _session, _flags_session, _socket_options
+    global _session, _flags_session, _capture_session, _socket_options
     if socket_options == _socket_options:
         return
     _socket_options = socket_options
     _session = _build_session(socket_options)
-    _flags_session = _build_flags_session(socket_options)
+    _flags_session = _build_no_retry_session(socket_options)
+    _capture_session = _build_no_retry_session(socket_options)
 
 
 def enable_keep_alive() -> None:
@@ -440,4 +451,10 @@ def default(self, obj: Any):
         return json.JSONEncoder.default(self, obj)
 
     def encode(self, o: Any) -> str:
+        # A strict encode fails only on a non-finite float, so finite events skip the copy.
+        if not self.allow_nan:
+            try:
+                return super().encode(o)
+            except ValueError:
+                pass
         return super().encode(_replace_non_finite_floats(o))

posthog/test/logging_helpers.py

@@ -1,7 +1,7 @@
 import io
 import logging
 from contextlib import contextmanager
-from typing import Iterator
+from typing import Iterator, Sequence
 
 from posthog._logging import _POSTHOG_LOGGER_NAME
 
@@ -25,3 +25,18 @@ def capture_message_only_logs(level: int = logging.DEBUG) -> Iterator[io.StringI
         handler.close()
         logger.setLevel(previous_level)
         logger.propagate = previous_propagate
+
+
+def assert_before_send_failure_logged(
+    records: Sequence[logging.LogRecord], event_name: str, error_type: str
+) -> None:
+    errors = [r for r in records if r.levelno >= logging.ERROR]
+    assert [r.getMessage() for r in errors] == [
+        f"[PostHog] Error in before_send callback for event {event_name} "
+        f"({error_type}), dropping the event"
+    ]
+    assert errors[0].exc_info is None
+    tracebacks = [r for r in records if r.exc_info]
+    assert [(r.levelno, r.exc_info[0].__name__) for r in tracebacks] == [
+        (logging.DEBUG, error_type)
+    ]

posthog/test/test_async_client.py

@@ -28,6 +28,7 @@
 )
 from posthog.request import APIError
 from posthog.test.capture_helpers import patch_async_capture_send
+from posthog.test.logging_helpers import assert_before_send_failure_logged
 
 
 @pytest.mark.asyncio
@@ -294,19 +295,37 @@ def before_send(event):
 
 
 @pytest.mark.asyncio
-async def test_capture_drops_event_when_before_send_raises():
+async def test_capture_drops_event_when_before_send_raises(caplog):
     async def before_send(_event):
         raise RuntimeError("callback failed")
 
     with patch_async_capture_send(new=mock.AsyncMock()) as send_batch:
         async with AsyncPosthog(
             "test-key", before_send=before_send, flush_interval=0.01
         ) as client:
-            accepted_uuid = client.capture("event", distinct_id="user-1")
-            await client.flush(timeout_seconds=1)
+            with caplog.at_level(logging.DEBUG, logger="posthog"):
+                accepted_uuid = client.capture("event", distinct_id="user-1")
+                await client.flush(timeout_seconds=1)
 
     assert accepted_uuid is not None
     send_batch.assert_not_awaited()
+    assert_before_send_failure_logged(caplog.records, "event", "RuntimeError")
+
+
+@pytest.mark.asyncio
+async def test_capture_drops_event_when_before_send_returns_non_dict(caplog):
+    with patch_async_capture_send(new=mock.AsyncMock()) as send_batch:
+        async with AsyncPosthog(
+            "test-key", before_send=lambda _event: "invalid", flush_interval=0.01
+        ) as client:
+            with caplog.at_level(logging.ERROR, logger="posthog"):
+                client.capture("event", distinct_id="user-1")
+                await client.flush(timeout_seconds=1)
+
+    send_batch.assert_not_awaited()
+    assert [r.getMessage() for r in caplog.records] == [
+        "[PostHog] before_send must return a dict or None, not str, dropping event event"
+    ]
 
 
 @pytest.mark.asyncio

posthog/test/test_before_send.py

@@ -7,6 +7,7 @@
 import posthog
 
 from posthog.client import Client
+from posthog.test.logging_helpers import assert_before_send_failure_logged
 from posthog.test.test_utils import FAKE_TEST_API_KEY
 from posthog.test.capture_helpers import (
     patch_capture_send,
@@ -172,14 +173,16 @@ def buggy_before_send(event):
                     "_normalize_event_uuid",
                     wraps=client._normalize_event_uuid,
                 ) as normalize_uuid,
-                self.assertLogs("posthog", level="ERROR") as logs,
+                self.assertLogs("posthog", level="DEBUG") as logs,
             ):
                 msg_uuid = client.capture("robust_event", distinct_id="user1")
 
             self.assertIsNone(msg_uuid)
             mock_post.assert_not_called()
             normalize_uuid.assert_called_once()
-            self.assertIn("Error in before_send callback: Oops!", logs.output[0])
+            assert_before_send_failure_logged(
+                logs.records, "robust_event", "ValueError"
+            )
 
     def test_before_send_callback_exception_does_not_enqueue_mutated_event(self):
         def buggy_before_send(event):
@@ -229,20 +232,22 @@ def add_unsupported_value(event):
         self.assertIsNone(sent_event["properties"]["marker"])
 
     def test_before_send_callback_non_dict_output_drops_event(self):
-        with (
-            patch_capture_send("client") as mock_post,
-            mock.patch("posthog.client.Client.log.exception") as mock_log,
-        ):
+        with patch_capture_send("client") as mock_post:
             client = Client(
                 FAKE_TEST_API_KEY,
                 before_send=lambda _event: "invalid",
                 sync_mode=True,
             )
-            self.assertIsNone(client.capture("original", distinct_id="user1"))
+            with self.assertLogs("posthog", level="ERROR") as logs:
+                self.assertIsNone(client.capture("original", distinct_id="user1"))
 
         mock_post.assert_not_called()
-        self.assertIn(
-            "before_send must return a dict or None", mock_log.call_args.args[0]
+        self.assertEqual(
+            logs.output,
+            [
+                "ERROR:posthog:[PostHog] before_send must return a dict or None, "
+                "not str, dropping event original"
+            ],
         )
 
     @parameterized.expand(

posthog/test/test_capture_invariants.py

@@ -47,7 +47,7 @@ def post(url, data=None, **kwargs):
 
     session = mock.Mock()
     session.post.side_effect = post
-    return mock.patch("posthog.capture_send._get_session", return_value=session)
+    return mock.patch("posthog.capture_send._get_capture_session", return_value=session)
 
 
 def _send_sync(method_name: str, before_send) -> list[tuple[str, list[dict]]]:

posthog/test/test_capture_send.py

@@ -7,6 +7,8 @@
 
 import zstandard
 
+from requests.adapters import HTTPAdapter
+
 from parameterized import parameterized
 
 from posthog import capture_send
@@ -732,6 +734,33 @@ def test_transport_error_exhausted_raises_capture_error(self) -> None:
         self.assertEqual(ctx.exception.attempts, 2)
         self.assertIsInstance(ctx.exception.__cause__, ConnectionError)
 
+    def test_refused_connection_is_retried_by_the_sdk_not_the_http_adapter(
+        self,
+    ) -> None:
+        # Each physical try must be its own SDK attempt, so capture gets a fresh request timestamp.
+        sent_attempts = []
+        real_send = HTTPAdapter.send
+
+        def record_send(adapter, request, **kwargs):
+            sent_attempts.append(request.headers[_HEADER_ATTEMPT])
+            return real_send(adapter, request, **kwargs)
+
+        with (
+            mock.patch.object(HTTPAdapter, "send", record_send),
+            mock.patch(
+                "urllib3.util.connection.create_connection",
+                side_effect=ConnectionRefusedError,
+            ) as connect,
+            self.assertRaises(CaptureError) as ctx,
+        ):
+            _send_v1_batch(
+                "phc_key", "https://app.posthog.com", [_msg("u-1")], max_retries=2
+            )
+
+        self.assertEqual(sent_attempts, ["1", "2", "3"])
+        self.assertEqual(connect.call_count, 3)
+        self.assertEqual(ctx.exception.attempts, 3)
+
     def test_negative_max_retries_still_attempts_delivery_once(self) -> None:
         stub = _PostV1Stub([_results_response({"u-1": "ok"})])