fix!: drop only the invalid event, not its whole batch

.sampo/changesets/invalid-events-and-sizes.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: major
+---
+
+An invalid event no longer costs the rest of its batch. An event with a missing, empty or non-string name, or with a non-string `distinct_id` after `before_send`, is dropped locally, an event that cannot be serialized (for example with `NaN`) is dropped alone, and a property key JSON cannot represent is dropped from its event. The analytics per-event limit is now capture's 10 MiB request limit instead of 900 KiB. A personal API key (`phx_...`) disables the client like an empty key. The SDK's own log lines no longer include event content or invalid timestamp values. `AsyncPosthog` follows the same rules.

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

@@ -23,6 +23,8 @@ You need to change code if your app does any of these:
 - tests the AI integrations with a mock client and asserts on `capture`
 - passes its own client object to the AI integrations
 - sets `$lib` or `$lib_version`, or filters on `$lib = "posthog-python-mcp"`
+- passes a personal API key (`phx_...`) as the project API key
+- reads event content from the SDK's debug log
 
 ## Endpoints
 
@@ -98,6 +100,19 @@ When the whole request is over the billing limit, capture returns a `402` with n
 - 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.
 - Retries start at 100 ms and double each attempt, up to 30 seconds. 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`.
+- 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
+
+Capture rejects a whole batch when one event cannot be parsed.
+8.0 drops only the invalid event, so the rest of the batch is delivered.
+
+- An event whose name is missing, `None`, empty or not a string is dropped. The SDK logs one warning per reason per process. 7.x sent it.
+- An event whose `before_send` hook sets `distinct_id` to `None` or a non-string is dropped, with one warning per reason per process. The SDK does not convert the hook's value or make up an ID. 7.x sent it, and capture would reject the whole batch.
+- PostHog drops an event whose name is over 200 UTF-8 bytes (`event_name_too_long`), and `on_error` reports it. 7.x sent events to legacy capture, which had no event name limit.
+- An event that cannot be serialized to JSON is dropped and logged, for example one with `NaN` or `Infinity` in its properties. 7.x sent `NaN` as invalid JSON.
+- A property or option key that JSON cannot represent, such as a tuple, is dropped without a log, and the rest of the event is sent. 7.x dropped the event. `int`, finite `float`, `bool` and `None` keys are still converted to strings. A `NaN` or infinite `float` key drops the event, as a `NaN` value does.
 
 ## Event uuids
 
@@ -196,7 +211,7 @@ These changes follow from this order:
 
 ## Batching and size limits
 
-- An analytics event over 900 KiB is dropped and logged. This now applies in `sync_mode` too.
+- An analytics event over 10 MiB is dropped and logged, because no capture request can carry it. 7.x dropped events over 900 KiB. Capture drops an event over its own per-event limit and reports it in `CaptureError.drops` as `event_too_big`.
 - An AI event over `capture_ai_max_event_bytes` is dropped and logged.
 - A batch stops before an event that would take it past 5 MiB. A larger event is sent alone.
 - `flush()` and `shutdown()` drain the analytics and AI queues at the same time.

posthog/__init__.py

@@ -12,6 +12,7 @@
 from posthog.capture_compression import CaptureCompression as CaptureCompression
 from posthog.capture_send import CaptureError as CaptureError
 from posthog.client import Client
+from posthog.utils import _is_personal_api_key
 from posthog.tracing.span import Span as Span
 from posthog.async_client import AsyncClient as AsyncClient
 from posthog.async_client import AsyncPosthog as AsyncPosthog
@@ -1497,8 +1498,12 @@ def setup() -> Client:
         )
 
     # Always set in case user changes it. Preserve Client's auto-disabled state
-    # for API keys that become empty after trimming.
-    default_client.disabled = disabled or not default_client.api_key
+    # for API keys that are empty after trimming or personal (`phx_`).
+    default_client.disabled = (
+        disabled
+        or not default_client.api_key
+        or _is_personal_api_key(default_client.api_key)
+    )
     default_client.debug = debug
     default_client.privacy_mode = bool(privacy_mode)
     default_client._set_before_send(before_send)

posthog/_async_consumer.py

@@ -78,7 +78,9 @@ async def _report_capture_failure(
 
 
 async def _serialized_event_size(event: dict[str, Any]) -> int:
-    serialized = await asyncio.to_thread(json.dumps, event, cls=DatetimeSerializer)
+    serialized = await asyncio.to_thread(
+        json.dumps, event, cls=DatetimeSerializer, allow_nan=False
+    )
     return len(serialized.encode())
 
 

posthog/async_client.py

@@ -98,8 +98,11 @@
 from .types import FlagMetadata, FlagValue, normalize_flags_response
 from .utils import (
     SizeLimitedDict,
+    _is_personal_api_key,
     _normalize_timestamp,
     clean,
+    _drop_invalid_distinct_id,
+    _drop_invalid_event_name,
     system_context,
 )
 from .version import VERSION
@@ -189,7 +192,9 @@ def __init__(
         self.on_error = on_error
         self.max_retries = max(0, max_retries)
         self.timeout = timeout
-        self.disabled = disabled or not self.api_key
+        self.disabled = (
+            disabled or not self.api_key or _is_personal_api_key(self.api_key)
+        )
         self.disable_geoip = disable_geoip
         self.is_server = is_server
         self.historical_migration = historical_migration
@@ -276,6 +281,11 @@ def __init__(
             self.log.error(
                 "api_key is empty after trimming whitespace; check your project API key"
             )
+        elif _is_personal_api_key(self.api_key):
+            self.log.error(
+                "api_key is a personal API key (phx_...), not a project API key; "
+                "PostHog is disabled. Use the project API key (phc_...)."
+            )
 
         self._analytics_lane = _AsyncLane(
             name="analytics",
@@ -472,8 +482,8 @@ def _prepare_event(
             msg["timestamp"] = _normalize_timestamp(timestamp)
         except ValueError:
             self.log.warning(
-                "Invalid timestamp %r. Falling back to the current UTC time.",
-                timestamp,
+                "Invalid timestamp; it must be a datetime or an ISO 8601 string. "
+                "Falling back to the current UTC time."
             )
             msg["timestamp"] = datetime.now(tz=timezone.utc).isoformat()
 
@@ -498,7 +508,14 @@ async def _process_event(
         self, msg: dict[str, Any], defaults: Optional[_EventDefaults]
     ) -> Optional[dict[str, Any]]:
         _fill_event_defaults(msg, defaults)
-        return await self._run_before_send(msg)
+        processed = await self._run_before_send(msg)
+        if (
+            processed is None
+            or _drop_invalid_event_name(self.log, processed)
+            or _drop_invalid_distinct_id(self.log, processed)
+        ):
+            return None
+        return processed
 
     async def _run_before_send(self, msg: dict[str, Any]) -> Optional[dict[str, Any]]:
         if self.before_send is None:

posthog/capture_send.py

@@ -342,7 +342,9 @@ def _post_v1(
     """
     trimmed_host = remove_trailing_slash(normalize_host(host))
     url = trimmed_host + path
-    data = json.dumps(batch_body, cls=DatetimeSerializer)
+    # Every event was serialized on its own with allow_nan=False before it
+    # joined the batch, so this never raises for an event the SDK accepted.
+    data = json.dumps(batch_body, cls=DatetimeSerializer, allow_nan=False)
     headers = {
         "Content-Type": "application/json",
         "User-Agent": sdk_info,

posthog/client.py

@@ -137,6 +137,9 @@
     SizeLimitedDict,
     clean,
     _normalize_timestamp,
+    _is_personal_api_key,
+    _drop_invalid_distinct_id,
+    _drop_invalid_event_name,
     guess_timezone as guess_timezone,
     system_context,
 )
@@ -997,7 +1000,9 @@ def __init__(
             _BackgroundEventLoopRunner
         ] = None
         self._flag_definition_cache_provider_async_runner_lock = threading.Lock()
-        self.disabled = disabled or not self.api_key
+        self.disabled = (
+            disabled or not self.api_key or _is_personal_api_key(self.api_key)
+        )
         self.disable_geoip = disable_geoip
         self._metrics_config = metrics
         self._metrics: Optional[PostHogMetrics] = None
@@ -1123,6 +1128,11 @@ def __init__(
             self.log.error(
                 "api_key is empty after trimming whitespace; check your project API key"
             )
+        elif _is_personal_api_key(self.api_key):
+            self.log.error(
+                "api_key is a personal API key (phx_...), not a project API key; "
+                "PostHog is disabled. Use the project API key (phc_...)."
+            )
 
         self._set_before_send(before_send)
 
@@ -2517,7 +2527,8 @@ def _enqueue(
             msg["timestamp"] = _normalize_timestamp(timestamp)
         except ValueError:
             self.log.warning(
-                "Invalid timestamp %r. Falling back to the current UTC time.", timestamp
+                "Invalid timestamp; it must be a datetime or an ISO 8601 string. "
+                "Falling back to the current UTC time."
             )
             msg["timestamp"] = datetime.now(tz=timezone.utc).isoformat()
 
@@ -2577,7 +2588,12 @@ def _enqueue(
         self._normalize_event_uuid(msg)
         sent_uuid = msg["uuid"]
 
-        self.log.debug("queueing: %s", msg)
+        if _drop_invalid_event_name(self.log, msg):
+            return None
+        if _drop_invalid_distinct_id(self.log, msg):
+            return None
+
+        self.log.debug("queueing %s on the %s lane", msg["event"], lane.name)
 
         # if send is False, return msg as if it was successfully queued, unless
         # shutdown has already closed this lane's admission.
@@ -2595,7 +2611,9 @@ def _enqueue(
             self.log.debug("enqueued with blocking %s.", msg["event"])
 
             try:
-                event_size = len(json.dumps(msg, cls=_DatetimeSerializer).encode())
+                event_size = len(
+                    json.dumps(msg, cls=_DatetimeSerializer, allow_nan=False).encode()
+                )
             except Exception:
                 self.log.error("Unable to serialize event for sizing, dropping.")
                 return None

posthog/consumer.py

@@ -19,7 +19,10 @@
 from queue import Empty
 
 
-MAX_MSG_SIZE = 900 * 1024  # 900KiB per event
+# Capture's request body limit. An analytics event larger than this can never
+# be accepted, so the SDK drops it locally; anything smaller is sent, and
+# capture reports an event over its own per-event limit in the response.
+MAX_MSG_SIZE = 10 * 1024 * 1024
 
 # The AI endpoint's per-event ceiling. The endpoint applies it to the
 # serialized properties alone.
@@ -33,8 +36,7 @@
 
 # A batch closes before it appends an event that would take it past this, so
 # a request carries at most this much event data, or one larger event alone.
-# The maximum request body size is currently 20MiB, let's be conservative
-# in case we want to lower it in the future.
+# Half of capture's request body limit, which applies before decompression.
 BATCH_SIZE_LIMIT = 5 * 1024 * 1024
 
 _configure_posthog_logging()
@@ -246,7 +248,9 @@ def next(self):
                     pending_items += 1
                     try:
                         item_size = len(
-                            json.dumps(item, cls=DatetimeSerializer).encode()
+                            json.dumps(
+                                item, cls=DatetimeSerializer, allow_nan=False
+                            ).encode()
                         )
                     except Exception:
                         # Callback-modified events can still contain invalid mapping

posthog/test/test_ai_capture_lane.py

@@ -227,10 +227,15 @@ def test_sync_mode_ai_event_at_endpoint_ceiling_is_sent(self):
         self.assertIsNotNone(result)
         mock_send.assert_called_once()
 
-    def test_analytics_lane_rejects_events_over_900kib(self):
+    def test_analytics_lane_sends_events_up_to_the_request_limit(self):
+        # Capture reports an analytics event over its per-event limit in the
+        # response, so the SDK only drops what no request could carry.
         client = self._client()
         consumer = client.consumers[0]
-        client.queue.put(self._sized_event("big_analytics_event", 2 * 1024 * 1024))
+        big = self._sized_event("big_analytics_event", 2 * 1024 * 1024)
+        client.queue.put(big)
+        client.queue.put(self._sized_event("too_big_for_a_request", MAX_MSG_SIZE))
+        self.assertEqual(consumer.next(), [big])
         self.assertEqual(consumer.next(), [])
         self.assertTrue(client.queue.empty())
 

posthog/test/test_async_client.py

@@ -204,6 +204,41 @@ async def send_batch(api_key, host, batch, **kwargs):
     assert batches[0][0]["uuid"] == event_uuid
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("bad_id", [123, 1.5, None, ["user"], {"id": "user"}])
+async def test_capture_drops_an_event_whose_before_send_distinct_id_is_not_a_string(
+    bad_id,
+):
+    batches = []
+
+    def before_send(event):
+        if event["event"] == "bad":
+            event["distinct_id"] = bad_id
+        return event
+
+    async def send_batch(api_key, host, batch, **kwargs):
+        batches.append(batch)
+
+    with (
+        patch_async_capture_send(side_effect=send_batch),
+        mock.patch("posthog.utils._warned_reasons", set()),
+        mock.patch.object(AsyncPosthog.log, "warning") as mock_warning,
+    ):
+        async with AsyncPosthog(
+            "test-key", before_send=before_send, flush_interval=30
+        ) as client:
+            client.capture("bad", distinct_id="user")
+            client.capture("good", distinct_id="user")
+            assert await client.capture_immediate("bad", distinct_id="user") is None
+            await client.flush(timeout_seconds=1)
+
+    events = [event for batch in batches for event in batch]
+    assert [(e["event"], e["distinct_id"]) for e in events] == [("good", "user")]
+    mock_warning.assert_called_once()
+    assert "distinct IDs must be strings" in mock_warning.call_args[0][0]
+    assert repr(bad_id) not in repr(mock_warning.call_args)
+
+
 @pytest.mark.asyncio
 async def test_capture_preserves_context_for_each_buffered_event():
     request_id = contextvars.ContextVar("request_id")
@@ -408,8 +443,23 @@ async def test_ai_capture_handles_a_non_string_event_name_like_capture(event):
     try:
         expected = client.capture(event, distinct_id="user-1") is None
         assert (client.capture_ai(event, distinct_id="user-1") is None) is expected
-        immediate = await client.capture_ai_immediate(event, distinct_id="user-1")
-        assert (immediate is None) is expected
+        # The immediate paths process the event, so they drop the invalid name.
+        assert await client.capture_immediate(event, distinct_id="user-1") is None
+        assert await client.capture_ai_immediate(event, distinct_id="user-1") is None
+    finally:
+        await client.shutdown()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("api_key", ["", " ", "phx_secret"])
+async def test_missing_or_personal_api_key_disables_the_client(api_key):
+    with mock.patch.object(AsyncPosthog.log, "error") as mock_error:
+        client = AsyncPosthog(api_key, send=False)
+    try:
+        assert client.disabled
+        assert client.capture("event", distinct_id="user-1") is None
+        mock_error.assert_called_once()
+        assert "secret" not in repr(mock_error.call_args)
     finally:
         await client.shutdown()
 

posthog/test/test_before_send.py

@@ -2,6 +2,8 @@
 import unittest
 from unittest import mock
 
+from parameterized import parameterized
+
 import posthog
 
 from posthog.client import Client
@@ -243,28 +245,45 @@ def test_before_send_callback_non_dict_output_drops_event(self):
             "before_send must return a dict or None", mock_log.call_args.args[0]
         )
 
-    def test_malformed_before_send_event_does_not_stop_consumer_or_shutdown(self):
-        def add_invalid_mapping_key(event):
+    @parameterized.expand(
+        [
+            # A key json cannot write costs only that key.
+            (
+                "unwritable_key",
+                ("private-key",),
+                "private-value",
+                ["malformed", "valid"],
+            ),
+            # A value json cannot write costs only that event, never the batch.
+            ("non_finite_value", "private-key", float("nan"), ["valid"]),
+        ]
+    )
+    def test_malformed_before_send_event_does_not_stop_consumer_or_shutdown(
+        self, _name, key, value, expected_events
+    ):
+        def add_malformed_property(event):
             if event["event"] == "malformed":
-                event["properties"][("private-key",)] = "private-value"
+                event["properties"][key] = value
             return event
 
         client = Client(
             FAKE_TEST_API_KEY,
-            before_send=add_invalid_mapping_key,
-            flush_at=1,
+            before_send=add_malformed_property,
+            flush_at=2,
             flush_interval=0.01,
         )
         with (
             patch_capture_send("consumer") as mock_post,
-            self.assertLogs("posthog", level="ERROR") as logs,
+            self.assertLogs("posthog", level="DEBUG") as logs,
         ):
             client.capture("malformed", distinct_id="user1")
             client.capture("valid", distinct_id="user1")
             client.shutdown()
 
         mock_post.assert_called_once()
-        self.assertEqual([event["event"] for event in sent_batch(mock_post)], ["valid"])
+        sent = sent_batch(mock_post)
+        self.assertEqual([event["event"] for event in sent], expected_events)
+        self.assertTrue(all(key not in event["properties"] for event in sent))
         self.assertEqual(client.queue.unfinished_tasks, 0)
         self.assertTrue(all(not consumer.is_alive() for consumer in client.consumers))
         self.assertNotIn("private-key", "\n".join(logs.output))