fix(email): stop blocking inline email sends on an analytics flush

posthog/email.py

@@ -546,6 +546,12 @@
     if not is_http_email_service_available():
         raise Exception("Missing Customer.io API key")
 
+    if not current_task or current_task.request.is_eager:
+        # The dedicated client blocks on its flush, which would hold the request. The web process keeps running,
+        # so the global client delivers in the background. ScopedCapture keeps the cloud-only guard.
+        _send_via_http_batch(ScopedCapture(posthoganalytics), to, campaign_key, template_name, properties, will_retry)
+        return
+
     # A dedicated client, because the global one may never flush before a Celery worker exits.
     with ph_scoped_capture(event_region=get_instance_region()) as capture:
         _send_via_http_batch(capture, to, campaign_key, template_name, properties, will_retry)
@@ -624,7 +630,7 @@
             raise
 
 
 def _send_via_smtp(
     to: list[dict[str, str]],
     campaign_key: str,
     subject: str,
@@ -779,7 +785,7 @@
     return "default"
 
 
 def sanitize_email_properties(properties: dict[str, Any] | None) -> dict[str, Any]:
     """
     Sanitizes properties that will be used in email templates to prevent HTML
     injection and to defang URL-shaped strings so mail clients don't auto-link
@@ -808,7 +814,7 @@
 
     supported_types = (str, int, float, bool, type(None), Decimal, uuid.UUID, datetime.datetime, datetime.date)
 
     def sanitize_value(value: Any, *, key: Any = None) -> Any:
         policy = _classify_key(key) if key is not None else "default"
         if policy == "passthrough":
             return value

posthog/test/test_email.py

@@ -312,12 +312,12 @@ def test_send_via_http_failed_metric_counts_undelivered_recipients(self, mock_po
         self.assertEqual(after - before, 2.0)  # b failed + c never attempted; a succeeded
 
     @parameterized.expand([("attempts_left", True), ("last_attempt", False)])
-    @patch("posthog.email.ph_scoped_capture")
+    @override_settings(CLOUD_DEPLOYMENT="US")
+    @patch("posthog.email.posthoganalytics.capture")
     @patch("posthog.email.requests.post")
     def test_send_via_http_reraises_a_lost_database_connection_so_celery_retries(
-        self, _name: str, will_retry: bool, mock_post: MagicMock, mock_scoped_capture: MagicMock
+        self, _name: str, will_retry: bool, mock_post: MagicMock, mock_capture: MagicMock
     ) -> None:
-        mock_capture = mock_scoped_capture.return_value.__enter__.return_value
         lost_connection = OperationalError("could not translate host name to address")
 
         with (
@@ -343,10 +343,10 @@ def test_send_via_http_reraises_a_lost_database_connection_so_celery_retries(
             [("OperationalError", will_retry)],
         )
 
-    @patch("posthog.email.ph_scoped_capture")
+    @override_settings(CLOUD_DEPLOYMENT="US")
+    @patch("posthog.email.posthoganalytics.capture")
     @patch("posthog.email.requests.post")
-    def test_send_via_http_success(self, mock_post, mock_scoped_capture) -> None:
-        mock_capture = mock_scoped_capture.return_value.__enter__.return_value
+    def test_send_via_http_success(self, mock_post, mock_capture) -> None:
         mock_response = MagicMock()
         mock_response.status_code = 200
         mock_response.json.return_value = {"delivery_id": "test_delivery_id", "queued_at": 1604977406}
@@ -387,6 +387,45 @@ def test_send_via_http_success(self, mock_post, mock_scoped_capture) -> None:
                 },
             )
 
+    @parameterized.expand(
+        [
+            ("request_on_cloud", None, "US", False, 1),
+            ("request_self_hosted", None, None, False, 0),
+            ("eager_task_on_cloud", True, "US", False, 1),
+            ("worker_task", False, "US", True, 0),
+        ]
+    )
+    @patch("posthog.email.ph_scoped_capture")
+    @patch("posthog.email.posthoganalytics.capture")
+    @patch("posthog.email.requests.post")
+    def test_send_via_http_flushes_events_only_in_a_worker(
+        self,
+        _name: str,
+        task_is_eager: bool | None,
+        cloud_deployment: str | None,
+        expect_scoped_client: bool,
+        expected_global_captures: int,
+        mock_post: MagicMock,
+        mock_global_capture: MagicMock,
+        mock_scoped_capture: MagicMock,
+    ) -> None:
+        mock_post.return_value = MagicMock(status_code=200, json=MagicMock(return_value={"delivery_id": "d-1"}))
+        task = None if task_is_eager is None else MagicMock(request=MagicMock(is_eager=task_is_eager))
+
+        with (
+            self.settings(CUSTOMER_IO_API_KEY="test-key", CLOUD_DEPLOYMENT=cloud_deployment),
+            patch("posthog.email.current_task", task),
+        ):
+            _send_via_http(
+                to=[{"raw_email": "a@posthog.com"}],
+                campaign_key="flush_path",
+                template_name="2fa_enabled",
+                properties={},
+            )
+
+        self.assertEqual(mock_scoped_capture.called, expect_scoped_client)
+        self.assertEqual(mock_global_capture.call_count, expected_global_captures)
+
     @patch("posthog.email.requests.post")
     def test_send_via_http_handles_decimal_values(self, mock_post) -> None:
         mock_response = MagicMock()
@@ -543,7 +582,8 @@ def test_workflows_flag_sees_the_instance_region(
             ),
         ]
     )
-    @patch("posthog.email.ph_scoped_capture")
+    @override_settings(CLOUD_DEPLOYMENT="US")
+    @patch("posthog.email.posthoganalytics.capture")
     @patch("posthog.email.posthoganalytics.feature_enabled")
     @patch("posthog.email.requests.post")
     def test_send_via_http_routes_through_workflows_when_flag_matches(
@@ -556,9 +596,8 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
         expected_events: list[str],
         mock_post: MagicMock,
         mock_feature_enabled: MagicMock,
-        mock_scoped_capture: MagicMock,
+        mock_capture: MagicMock,
     ) -> None:
-        mock_capture = mock_scoped_capture.return_value.__enter__.return_value
         mock_feature_enabled.return_value = flag_enabled
         responses: list[MagicMock | Exception] = []
         for status_code in status_codes:
@@ -606,7 +645,11 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
         mock_feature_enabled.assert_called_once_with(
             "transactional-emails-via-workflows",
             "user-1",
-            person_properties={"email": "test@posthog.com", "transactional_email_template": "2fa_enabled"},
+            person_properties={
+                "email": "test@posthog.com",
+                "transactional_email_template": "2fa_enabled",
+                "region": "US",
+            },
             send_feature_flag_events=False,
         )
 
@@ -672,7 +715,7 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
             ),
         ]
     )
-    @patch("posthog.email.ph_scoped_capture")
+    @patch("posthog.email.posthoganalytics.capture")
     @patch("posthog.email.posthoganalytics.feature_enabled")
     @patch("posthog.email.requests.post")
     def test_retry_after_an_uncertain_workflows_send_stays_on_workflows_while_it_is_configured(
@@ -717,7 +760,7 @@ def test_retry_after_an_uncertain_workflows_send_stays_on_workflows_while_it_is_
         ]
         self.assertEqual(providers, expected_providers)
 
-    @patch("posthog.email.ph_scoped_capture")
+    @patch("posthog.email.posthoganalytics.capture")
     @patch("posthog.email.posthoganalytics.feature_enabled")
     @patch("posthog.email.requests.post")
     def test_retry_after_failing_to_mark_a_workflows_send_replays_it_on_workflows(
@@ -770,7 +813,7 @@ def lose_connection_on_first_sent_mark(record: MessagingRecord, *args: Any, **kw
         ]
     )
     @patch.object(_send_email, "apply_async")
-    @patch("posthog.email.ph_scoped_capture")
+    @patch("posthog.email.posthoganalytics.capture")
     @patch("posthog.email.posthoganalytics.feature_enabled", return_value=True)
     @patch("posthog.email.requests.post", side_effect=requests.ReadTimeout("no response"))
     def test_inline_send_hands_a_retryable_failure_to_a_worker(