fix(analytics): allow capability to offload reportExposure to async thread (SDK-80)

mixpanel/flags/local_feature_flags.py

@@ -18,8 +18,8 @@
     SelectedVariant,
 )
 from .utils import (
-    EXPOSURE_EVENT,
     REQUEST_HEADERS,
+    dispatch_exposure,
     generate_traceparent,
     normalized_hash,
     prepare_common_query_params,
@@ -511,12 +511,17 @@ def _track_exposure(
             if latency_in_seconds is not None:
                 properties["Variant fetch latency (ms)"] = latency_in_seconds * 1000
 
-            self._tracker(distinct_id, EXPOSURE_EVENT, properties)
+            self._dispatch_exposure(distinct_id, properties)
         else:
             logger.error(
                 "Cannot track exposure event without a distinct_id in the context"
             )
 
+    def _dispatch_exposure(self, distinct_id: str, properties: dict[str, Any]) -> None:
+        dispatch_exposure(
+            self._tracker, self._config.exposure_executor, distinct_id, properties
+        )
+
     async def __aenter__(self):
         return self
 

mixpanel/flags/remote_feature_flags.py

@@ -14,6 +14,7 @@
 from .utils import (
     EXPOSURE_EVENT,
     REQUEST_HEADERS,
+    dispatch_exposure,
     generate_traceparent,
     prepare_common_query_params,
 )
@@ -237,7 +238,7 @@ def get_variant(
                 properties = self._build_tracking_properties(
                     flag_key, selected_variant, start_time, end_time
                 )
-                self._tracker(distinct_id, EXPOSURE_EVENT, properties)
+                self._dispatch_exposure(distinct_id, properties)
 
         except Exception:
             logger.exception("Failed to get remote variant for flag '%s'", flag_key)
@@ -267,12 +268,17 @@ def track_exposure_event(
         """
         if distinct_id := context.get("distinct_id"):
             properties = self._build_tracking_properties(flag_key, variant)
-            self._tracker(distinct_id, EXPOSURE_EVENT, properties)
+            self._dispatch_exposure(distinct_id, properties)
         else:
             logger.error(
                 "Cannot track exposure event without a distinct_id in the context"
             )
 
+    def _dispatch_exposure(self, distinct_id: str, properties: dict[str, Any]) -> None:
+        dispatch_exposure(
+            self._tracker, self._config.exposure_executor, distinct_id, properties
+        )
+
     def _prepare_query_params(
         self, context: dict[str, Any], flag_key: str | None = None
     ) -> dict[str, str]:

mixpanel/flags/test_local_feature_flags.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import threading
+from concurrent.futures import ThreadPoolExecutor
 from itertools import chain, repeat
 from typing import Any
 from unittest.mock import Mock, patch
@@ -646,6 +647,83 @@ async def test_get_variant_value_does_not_track_exposure_without_distinct_id(sel
         )
         self._mock_tracker.assert_not_called()
 
+    @respx.mock
+    async def test_default_exposure_runs_inline_on_calling_thread(self):
+        """Smoke test: exposure_executor defaults to None, tracker runs inline."""
+        flag = create_test_flag(rollout_percentage=100.0)
+        await self.setup_flags([flag])
+
+        called_on: list[threading.Thread] = []
+
+        def tracker(_distinct_id, _event, _properties):
+            called_on.append(threading.current_thread())
+
+        self._mock_tracker.side_effect = tracker
+        self._flags.get_variant_value(TEST_FLAG_KEY, "fallback", USER_CONTEXT)
+        assert called_on == [threading.current_thread()]
+
+    @respx.mock
+    async def test_track_exposure_event_routes_through_executor(self):
+        """Manual API also honors exposure_executor."""
+        executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="exposure")
+        try:
+            tracker_done = threading.Event()
+            captured: list[threading.Thread] = []
+
+            def tracker(_distinct_id, _event, _properties):
+                captured.append(threading.current_thread())
+                tracker_done.set()
+
+            config = LocalFlagsConfig(enable_polling=False, exposure_executor=executor)
+            provider = LocalFeatureFlagsProvider(
+                "test-token", config, "1.0.0", Mock(side_effect=tracker)
+            )
+            try:
+                provider.track_exposure_event(
+                    "manual",
+                    SelectedVariant(variant_key="treatment", variant_value="x"),
+                    USER_CONTEXT,
+                )
+                assert tracker_done.wait(timeout=2.0)
+                assert captured[0] is not threading.current_thread()
+                assert captured[0].name.startswith("exposure")
+            finally:
+                await provider.__aexit__(None, None, None)
+        finally:
+            executor.shutdown(wait=True)
+
+    @respx.mock
+    async def test_exposure_executor_dispatches_tracker_off_calling_thread(self):
+        flag = create_test_flag(rollout_percentage=100.0)
+        executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="exposure")
+        try:
+            calling_thread = threading.current_thread()
+            tracker_thread = threading.Event()
+            captured_thread: list[threading.Thread] = []
+
+            def tracker(_distinct_id, _event, _properties):
+                captured_thread.append(threading.current_thread())
+                tracker_thread.set()
+
+            tracker_mock = Mock(side_effect=tracker)
+            config = LocalFlagsConfig(enable_polling=False, exposure_executor=executor)
+            provider = LocalFeatureFlagsProvider(
+                "test-token", config, "1.0.0", tracker_mock
+            )
+            try:
+                respx.get("https://api.mixpanel.com/flags/definitions").mock(
+                    return_value=create_flags_response([flag])
+                )
+                await provider.astart_polling_for_definitions()
+                provider.get_variant_value(TEST_FLAG_KEY, "fallback", USER_CONTEXT)
+                assert tracker_thread.wait(timeout=2.0), "tracker never ran"
+                assert captured_thread[0] is not calling_thread
+                assert captured_thread[0].name.startswith("exposure")
+            finally:
+                await provider.__aexit__(None, None, None)
+        finally:
+            executor.shutdown(wait=True)
+
     @respx.mock
     async def test_get_all_variants_returns_all_variants_when_user_in_rollout(self):
         flag1 = create_test_flag(flag_key="flag1", rollout_percentage=100.0)

mixpanel/flags/test_remote_feature_flags.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import asyncio
+import threading
+from concurrent.futures import ThreadPoolExecutor
 from unittest.mock import Mock
 
 import httpx
@@ -281,6 +283,96 @@ def test_get_variant_value_does_not_track_exposure_event_if_fallback(self):
         )
         self.mock_tracker.assert_not_called()
 
+    def test_default_exposure_runs_inline_on_calling_thread(self):
+        """Smoke test: exposure_executor defaults to None, tracker runs inline."""
+        called_on: list[threading.Thread] = []
+
+        def tracker(_distinct_id, _event, _properties):
+            called_on.append(threading.current_thread())
+
+        provider = RemoteFeatureFlagsProvider(
+            "test-token", RemoteFlagsConfig(), "1.0.0", tracker
+        )
+        try:
+            provider.track_exposure_event(
+                "manual",
+                SelectedVariant(variant_key="treatment", variant_value="x"),
+                {"distinct_id": "user123"},
+            )
+            assert called_on == [threading.current_thread()]
+        finally:
+            provider.__exit__(None, None, None)
+
+    def test_track_exposure_event_routes_through_executor(self):
+        """Manual API also honors exposure_executor."""
+        executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="exposure")
+        try:
+            tracker_done = threading.Event()
+            captured: list[threading.Thread] = []
+
+            def tracker(_distinct_id, _event, _properties):
+                captured.append(threading.current_thread())
+                tracker_done.set()
+
+            provider = RemoteFeatureFlagsProvider(
+                "test-token",
+                RemoteFlagsConfig(exposure_executor=executor),
+                "1.0.0",
+                Mock(side_effect=tracker),
+            )
+            try:
+                provider.track_exposure_event(
+                    "manual",
+                    SelectedVariant(variant_key="treatment", variant_value="x"),
+                    {"distinct_id": "user123"},
+                )
+                assert tracker_done.wait(timeout=2.0)
+                assert captured[0] is not threading.current_thread()
+                assert captured[0].name.startswith("exposure")
+            finally:
+                provider.__exit__(None, None, None)
+        finally:
+            executor.shutdown(wait=True)
+
+    @respx.mock
+    def test_exposure_executor_dispatches_tracker_off_calling_thread(self):
+        respx.get(ENDPOINT).mock(
+            return_value=create_success_response(
+                {
+                    "test_flag": SelectedVariant(
+                        variant_key="treatment", variant_value="treatment"
+                    )
+                }
+            )
+        )
+
+        executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="exposure")
+        try:
+            calling_thread = threading.current_thread()
+            tracker_thread = threading.Event()
+            captured_thread: list[threading.Thread] = []
+
+            def tracker(_distinct_id, _event, _properties):
+                captured_thread.append(threading.current_thread())
+                tracker_thread.set()
+
+            tracker_mock = Mock(side_effect=tracker)
+            config = RemoteFlagsConfig(exposure_executor=executor)
+            provider = RemoteFeatureFlagsProvider(
+                "test-token", config, "1.0.0", tracker_mock
+            )
+            try:
+                provider.get_variant_value(
+                    "test_flag", "control", {"distinct_id": "user123"}
+                )
+                assert tracker_thread.wait(timeout=2.0), "tracker never ran"
+                assert captured_thread[0] is not calling_thread
+                assert captured_thread[0].name.startswith("exposure")
+            finally:
+                provider.__exit__(None, None, None)
+        finally:
+            executor.shutdown(wait=True)
+
     @respx.mock
     def test_is_enabled_returns_true_for_true_variant_value(self):
         respx.get(ENDPOINT).mock(

mixpanel/flags/test_utils.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
+import logging
 import re
+from concurrent.futures import ThreadPoolExecutor
+from unittest.mock import MagicMock
 
 import pytest
 
-from .utils import generate_traceparent, normalized_hash
+from .utils import dispatch_exposure, generate_traceparent, normalized_hash
 
 
 class TestUtils:
@@ -31,3 +34,28 @@ def test_normalized_hash_for_known_inputs(self, key, salt, expected_hash):
         assert result == expected_hash, (
             f"Expected hash of {expected_hash} for '{key}' with salt '{salt}', got {result}"
         )
+
+    def test_dispatch_exposure_runs_inline_when_no_executor(self):
+        tracker = MagicMock()
+
+        dispatch_exposure(tracker, None, "user-1", {"prop": "value"})
+
+        tracker.assert_called_once_with("user-1", "$experiment_started", {"prop": "value"})
+
+    def test_dispatch_exposure_logs_executor_thread_exceptions(self, caplog):
+        # Without the done-callback the future.exception() would be silently
+        # discarded — this test would fail (no log record captured).
+        def boom(*_args, **_kwargs):
+            raise RuntimeError("tracker exploded")
+
+        with ThreadPoolExecutor(max_workers=1) as executor:
+            with caplog.at_level(logging.ERROR, logger="mixpanel.flags.utils"):
+                dispatch_exposure(boom, executor, "user-1", {})
+            # Drain the executor so the done-callback has a chance to fire.
+            executor.shutdown(wait=True)
+
+        assert any(
+            "Exposure event failed on executor thread" in rec.message
+            and "tracker exploded" in rec.message
+            for rec in caplog.records
+        ), f"expected error log, got {[r.message for r in caplog.records]}"

mixpanel/flags/types.py

@@ -1,3 +1,4 @@
+from concurrent.futures import Executor
 from typing import Any, Optional
 
 from pydantic import BaseModel, ConfigDict
@@ -10,6 +11,10 @@ class FlagsConfig(BaseModel):
 
     api_host: str = "api.mixpanel.com"
     request_timeout_in_seconds: int = 10
+    # Optional executor used to dispatch exposure-event HTTP sends so flag
+    # evaluation does not block on the network round trip. None (default)
+    # preserves the existing inline behavior.
+    exposure_executor: Optional[Executor] = None
 
 
 class LocalFlagsConfig(FlagsConfig):

mixpanel/flags/utils.py

@@ -1,6 +1,11 @@
 from __future__ import annotations
 
+import logging
 import uuid
+from concurrent.futures import Executor, Future
+from typing import Any, Callable
+
+logger = logging.getLogger(__name__)
 
 EXPOSURE_EVENT = "$experiment_started"
 
@@ -49,6 +54,44 @@ def prepare_common_query_params(token: str, sdk_version: str) -> dict[str, str]:
     return {"mp_lib": "python", "lib_version": sdk_version, "token": token}
 
 
+def dispatch_exposure(
+    tracker: Callable,
+    executor: Executor | None,
+    distinct_id: str,
+    properties: dict[str, Any],
+) -> None:
+    """Invoke the tracker inline or via the configured executor.
+
+    Shared between local and remote flag providers so a change to the
+    dispatch policy only needs to happen in one place.
+    """
+    if executor is None:
+        tracker(distinct_id, EXPOSURE_EVENT, properties)
+        return
+
+    try:
+        future = executor.submit(tracker, distinct_id, EXPOSURE_EVENT, properties)
+    except RuntimeError:
+        logger.exception(
+            "Exposure event dropped — executor refused to accept task"
+        )
+        return
+
+    # Retrieve exceptions raised on the executor thread; otherwise a
+    # tracker raising on the background thread is swallowed by the Future.
+    future.add_done_callback(_log_tracker_future_exception)
+
+
+def _log_tracker_future_exception(future: Future) -> None:
+    exc = future.exception()
+    if exc is not None:
+        logger.error(
+            "Exposure event failed on executor thread: %s: %s",
+            type(exc).__name__,
+            exc,
+        )
+
+
 def generate_traceparent() -> str:
     """Generate a W3C traceparent header for distributed tracing interop.