fix(capture): handle malformed before_send results

.sampo/changesets/valorous-witch-ukko.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: patch
+---
+
+Keep consumers alive after malformed before_send results

posthog/client.py

@@ -1946,7 +1946,9 @@ def _enqueue(self, msg, disable_geoip, lane=None, property_allowlist=None):
                 if modified_msg is None:
                     self.log.debug("Event dropped by before_send callback")
                     return None
-                msg = modified_msg
+                if not isinstance(modified_msg, dict):
+                    raise TypeError("before_send must return a dict or None")
+                msg = clean(modified_msg)
             except Exception as e:
                 self.log.exception(f"Error in before_send callback: {e}")
                 # Continue with the original message if callback fails

posthog/consumer.py

@@ -132,7 +132,16 @@ def next(self):
                 break
             try:
                 item = queue.get(block=True, timeout=self.flush_interval - elapsed)
-                item_size = len(json.dumps(item, cls=DatetimeSerializer).encode())
+                try:
+                    item_size = len(json.dumps(item, cls=DatetimeSerializer).encode())
+                except Exception:
+                    # Callback-modified events can still contain invalid mapping
+                    # keys or circular references. Never log the payload here.
+                    self.log.error(
+                        "Unable to serialize queued event for sizing, dropping."
+                    )
+                    queue.task_done()
+                    continue
                 if item_size > self.max_msg_size:
                     # Log only name and size: AI events may carry unredacted
                     # multimodal payloads that must not leak into logs.

posthog/test/test_before_send.py

@@ -121,6 +121,70 @@ def buggy_before_send(event):
             enqueued_msg = batch_data[0]
             self.assertEqual(enqueued_msg["event"], "robust_event")
 
+    def test_before_send_callback_output_is_recleaned(self):
+        marker = object()
+
+        def add_unsupported_value(event):
+            event["properties"]["marker"] = marker
+            return event
+
+        with mock.patch("posthog.client.batch_post") as mock_post:
+            client = Client(
+                FAKE_TEST_API_KEY,
+                before_send=add_unsupported_value,
+                sync_mode=True,
+            )
+            self.assertIsNotNone(client.capture("recleaned", distinct_id="user1"))
+
+        sent_event = mock_post.call_args.kwargs["batch"][0]
+        self.assertIsNone(sent_event["properties"]["marker"])
+
+    def test_before_send_callback_non_dict_output_uses_original_event(self):
+        with (
+            mock.patch("posthog.client.batch_post") as mock_post,
+            mock.patch("posthog.client.Client.log.exception") as mock_log,
+        ):
+            client = Client(
+                FAKE_TEST_API_KEY,
+                before_send=lambda _event: "invalid",
+                sync_mode=True,
+            )
+            self.assertIsNotNone(client.capture("original", distinct_id="user1"))
+
+        sent_event = mock_post.call_args.kwargs["batch"][0]
+        self.assertEqual(sent_event["event"], "original")
+        self.assertIn(
+            "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):
+            if event["event"] == "malformed":
+                event["properties"][("private-key",)] = "private-value"
+            return event
+
+        client = Client(
+            FAKE_TEST_API_KEY,
+            before_send=add_invalid_mapping_key,
+            flush_at=1,
+            flush_interval=0.01,
+        )
+        with (
+            mock.patch("posthog.consumer.batch_post") as mock_post,
+            self.assertLogs("posthog", level="ERROR") as logs,
+        ):
+            client.capture("malformed", distinct_id="user1")
+            client.capture("valid", distinct_id="user1")
+            client.shutdown()
+
+        mock_post.assert_called_once()
+        sent_batch = mock_post.call_args.kwargs["batch"]
+        self.assertEqual([event["event"] for event in sent_batch], ["valid"])
+        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))
+        self.assertNotIn("private-value", "\n".join(logs.output))
+
     def test_before_send_callback_works_with_all_event_types(self):
         """Test that before_send works with capture, set, etc."""