fix(email): stop falling back to customer.io for transactional emails

posthog/email.py

@@ -108,7 +108,8 @@
     "ignore_result": True,
     "autoretry_for": (Exception,),
     "max_retries": 3,
-    # Retries wait 5s, 10s and 20s. Without jitter the last attempt cannot come within a second of the first.
+    # SMTP retries wait 5s, 10s and 20s. Without jitter the last attempt cannot come within a second of the first.
+    # HTTP retries use HTTP_SEND_RETRY_COUNTDOWNS_SECONDS instead.
     "retry_backoff": 5,
     "retry_jitter": False,
 }
@@ -220,20 +221,14 @@
 # Matches how long the send API keeps an idempotency key, which outlasts every Celery retry.
 WORKFLOWS_PIN_TTL_SECONDS = 24 * 60 * 60
 
-# 422 codes for an email the send API refused before queueing it: sending is suspended for the project or paused for
-# the workflow, or the recipient is on the suppression list. Customer.io sends these emails instead.
-WORKFLOWS_NOT_QUEUED_422_CODES = frozenset({"sending_suspended", "sending_paused", "recipient_suppressed"})
-
 # 422 codes for an email the send API refused before queueing it because the address can't receive email: the syntax
-# is invalid or the domain has no mail servers. Every provider would hard bounce it, so the email is dropped. It is not
-# retried and Customer.io does not send it.
-WORKFLOWS_UNDELIVERABLE_422_CODES = frozenset({"recipient_unreachable"})
-
-EMAIL_WORKFLOWS_FALLBACK_COUNTER = Counter(
-    "posthog_email_workflows_fallback_total",
-    "Emails that Workflows refused and Customer.io sent instead, by template.",
-    labelnames=["template_name"],
-)
+# is invalid, the domain has no mail servers, or the address hard bounced and is on the suppression list. Every
+# provider would bounce it, so the email is dropped. It is not retried and Customer.io does not send it.
+WORKFLOWS_UNDELIVERABLE_422_CODES = frozenset({"recipient_unreachable", "recipient_suppressed"})
+
+# 422 codes for an email the send API refused before queueing it: sending is suspended for the project or paused for
+# the workflow.
+WORKFLOWS_NOT_QUEUED_422_CODES = frozenset({"sending_suspended", "sending_paused"})
 
 
 def get_email_team_and_org_context(
@@ -288,14 +283,20 @@
         self.status_code = status_code
 
 
+# Waits between worker attempts of an HTTP send that failed with one of the errors below. Customer.io no longer sends
+# an email that Workflows could not take, so the retries span about 19 minutes to outlast a short Workflows outage.
+# They stop there, because a login code or a password reset link that arrives later is of no use to the recipient.
+# SMTP sends keep the shorter retries in EMAIL_TASK_KWARGS.
+HTTP_SEND_RETRY_COUNTDOWNS_SECONDS = (10, 30, 60, 120, 300, 600)
+
 # Errors that a later attempt can fix. A database error can come before the send, when the record is read, or after
 # it, when the record is marked sent. A Workflows send is pinned before that write, so its retry replays under the
 # same idempotency key. A Customer.io send can go out a second time.
 RETRYABLE_HTTP_SEND_ERRORS = (WorkflowsEmailError, OperationalError, InterfaceError)
 
 
-def _workflows_error_allows_fallback(err: WorkflowsEmailError) -> bool:
-    """Whether Customer.io can send the email without the risk of a duplicate."""
+def _workflows_queued_nothing(err: WorkflowsEmailError) -> bool:
+    """Whether the email is certain not to be queued, so a retry can pick its provider again."""
     if err.status_code is None:
         # A connection that was never made and a client that queued nothing sent nothing. Any other network
         # failure can come after Workflows queued the email.
@@ -516,8 +517,8 @@
     Returns None when Workflows refused the recipient as undeliverable, so no provider sent the email.
     """
     idempotency_key = _workflows_idempotency_key(campaign_key, dest["raw_email"])
-    # After an attempt Workflows may have queued, a retry skips the flag and never falls back. Otherwise a flag
-    # change or a different error on the retry would send the same email again through Customer.io.
+    # After an attempt Workflows may have queued, a retry skips the flag. Otherwise a flag change on the retry would
+    # send the same email again through Customer.io.
     pinned_template_id = _pinned_workflows_template_id(idempotency_key)
     workflows_template_id = get_workflows_template_id(template_name, dest, pinned_template_id)
     if workflows_template_id:
@@ -540,17 +541,11 @@
                     {"template_name": template_name, "campaign_key": campaign_key, "error": str(err)[:500]},
                 )
                 return None
-            if pinned_template_id or not _workflows_error_allows_fallback(err):
+            # Workflows sends the retry. An email it certainly did not queue stays unpinned, so turning the flag off
+            # moves the retry to Customer.io.
+            if pinned_template_id or not _workflows_queued_nothing(err):
                 _pin_to_workflows(idempotency_key, workflows_template_id)
-                raise
-            logger.warning("email_workflows_send_rejected", template_name=template_name, error=str(err))
-            EMAIL_WORKFLOWS_FALLBACK_COUNTER.labels(template_name=template_name).inc()
-            _capture_email_event(
-                capture,
-                dest,
-                "transactional email workflows rejected",
-                {"template_name": template_name, "campaign_key": campaign_key, "error": str(err)[:500]},
-            )
+            raise
     return "customer_io", _send_one_via_customer_io(dest, template_name, properties)
 
 
@@ -660,7 +655,7 @@
             raise
 
 
 def _send_via_smtp(
     to: list[dict[str, str]],
     campaign_key: str,
     subject: str,
@@ -815,7 +810,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
@@ -844,7 +839,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
@@ -880,14 +875,20 @@
     return {k: sanitize_value(v, key=k) for k, v in properties.items()}
 
 
+def _worker_send_task() -> Any:
+    """The `_send_email` task that a worker runs this send in, or None for an inline or eager send."""
+    task = current_task
+    if not task or task.name != "posthog.email._send_email" or task.request.is_eager:
+        return None
+    return task
+
+
 def _send_will_retry(retry_in_worker: bool) -> bool:
     """Whether a retryable failure of this send gets another attempt."""
     if retry_in_worker:
         return True
-    task = current_task
-    if not task or task.name != "posthog.email._send_email" or task.request.is_eager:
-        return False
-    return task.request.retries < task.max_retries
+    task = _worker_send_task()
+    return task is not None and task.request.retries < len(HTTP_SEND_RETRY_COUNTDOWNS_SECONDS)
 
 
 def _send_email_now(
@@ -912,8 +913,17 @@
                 properties=properties or {},
                 will_retry=_send_will_retry(retry_in_worker),
             )
-        except RETRYABLE_HTTP_SEND_ERRORS:
+        except RETRYABLE_HTTP_SEND_ERRORS as err:
             if not retry_in_worker:
+                task = _worker_send_task()
+                if task is not None:
+                    # After the last wait, Celery raises `err` instead of scheduling another attempt.
+                    countdowns = HTTP_SEND_RETRY_COUNTDOWNS_SECONDS
+                    raise task.retry(
+                        exc=err,
+                        countdown=countdowns[min(task.request.retries, len(countdowns) - 1)],
+                        max_retries=len(countdowns),
+                    )
                 raise
             # The retry sends the same idempotency key, so Workflows does not send the email twice.
             _send_email.apply_async(

posthog/test/test_email.py

@@ -22,11 +22,13 @@
 
 import urllib3
 import requests
+from celery.exceptions import Retry
 from parameterized import parameterized
 from prometheus_client import REGISTRY
 
 from posthog.email import (
     CUSTOMER_IO_TEMPLATE_ID_MAP,
+    HTTP_SEND_RETRY_COUNTDOWNS_SECONDS,
     WORKFLOWS_TEMPLATE_ID_MAP,
     EmailMessage,
     WorkflowsEmailError,
@@ -505,22 +507,23 @@ def test_workflows_flag_sees_the_instance_region(
     @parameterized.expand(
         [
             ("workflows_accepts", True, [202], ["workflows"], "workflows", ["transactional email triggered"]),
+            ("workflows_rejects_sends_nothing", True, [400], ["workflows"], None, ["transactional email failed"]),
+            ("workflows_in_flight_sends_nothing", True, [409], ["workflows"], None, ["transactional email failed"]),
             (
-                "workflows_rejects_falls_back",
+                "workflows_refuses_suppressed_recipient_drops_email",
                 True,
-                [400, 200],
-                ["workflows", "customer_io"],
-                "customer_io",
-                ["transactional email workflows rejected", "transactional email triggered"],
+                [(422, {"type": "invalid_request", "code": "recipient_suppressed"})],
+                ["workflows"],
+                None,
+                ["transactional email workflows rejected"],
             ),
-            ("workflows_in_flight_sends_nothing", True, [409], ["workflows"], None, ["transactional email failed"]),
             (
-                "workflows_refuses_suppressed_recipient_falls_back",
+                "workflows_refuses_paused_sending_sends_nothing",
                 True,
-                [(422, {"type": "invalid_request", "code": "recipient_suppressed"}), 200],
-                ["workflows", "customer_io"],
-                "customer_io",
-                ["transactional email workflows rejected", "transactional email triggered"],
+                [(422, {"type": "invalid_request", "code": "sending_paused"})],
+                ["workflows"],
+                None,
+                ["transactional email failed"],
             ),
             (
                 "workflows_refuses_unreachable_recipient_drops_email",
@@ -548,7 +551,7 @@ def test_workflows_flag_sees_the_instance_region(
                 ["transactional email failed"],
             ),
             (
-                "workflows_unreachable_falls_back",
+                "workflows_unreachable_sends_nothing",
                 True,
                 [
                     requests.ConnectionError(
@@ -557,12 +560,11 @@ def test_workflows_flag_sees_the_instance_region(
                             "/",
                             urllib3.exceptions.NewConnectionError(None, "refused"),
                         )
-                    ),
-                    200,
+                    )
                 ],
-                ["workflows", "customer_io"],
-                "customer_io",
-                ["transactional email workflows rejected", "transactional email triggered"],
+                ["workflows"],
+                None,
+                ["transactional email failed"],
             ),
             (
                 "workflows_reset_after_send_sends_nothing",
@@ -695,13 +697,15 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
         [
             (
                 "flag_turned_off",
+                requests.ReadTimeout("no response"),
                 False,
                 "phs_test",
                 MagicMock(status_code=202, json=MagicMock(return_value={"message_id": "m1"})),
                 ["workflows", "workflows"],
             ),
             (
                 "workflows_unreachable",
+                requests.ReadTimeout("no response"),
                 True,
                 "phs_test",
                 requests.ConnectionError(
@@ -716,19 +720,30 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
             # Removing the key turns Workflows off, so a pinned retry must not post the email without it.
             (
                 "api_key_removed",
+                requests.ReadTimeout("no response"),
                 True,
                 "",
                 MagicMock(status_code=200, json=MagicMock(return_value={"delivery_id": "d1"})),
                 ["workflows", "customer_io"],
             ),
+            # Workflows queued nothing, so turning the flag off moves the retry to Customer.io.
+            (
+                "refused_before_queueing_then_flag_turned_off",
+                MagicMock(status_code=400, text="error", json=MagicMock(return_value={"code": "invalid_input"})),
+                False,
+                "phs_test",
+                MagicMock(status_code=200, json=MagicMock(return_value={"delivery_id": "d1"})),
+                ["workflows", "customer_io"],
+            ),
         ]
     )
     @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(
+    def test_retry_after_a_failed_workflows_send_stays_on_workflows_unless_nothing_was_queued(
         self,
         _name: str,
+        first_response: MagicMock | Exception,
         flag_on_retry: bool,
         api_key_on_retry: str,
         retry_response: MagicMock | Exception,
@@ -738,7 +753,7 @@ def test_retry_after_an_uncertain_workflows_send_stays_on_workflows_while_it_is_
         _mock_capture: MagicMock,
     ) -> None:
         mock_feature_enabled.side_effect = [True, flag_on_retry]
-        mock_post.side_effect = [requests.ReadTimeout("no response"), retry_response]
+        mock_post.side_effect = [first_response, retry_response]
 
         with (
             override_instance_config("EMAIL_HOST", "localhost"),
@@ -899,6 +914,60 @@ def test_inline_send_hands_a_retryable_failure_to_a_worker(
         # A worker still on the previous release rejects an argument it does not know.
         self.assertNotIn("retry_in_worker", retry_kwargs)
 
+    @parameterized.expand(
+        [
+            ("first_attempt_waits_10s", 0, 10, True),
+            ("last_attempt_gives_up", len(HTTP_SEND_RETRY_COUNTDOWNS_SECONDS), None, False),
+        ]
+    )
+    @patch.object(_send_email, "apply_async")
+    @patch("posthog.email.ph_scoped_capture")
+    @patch("posthog.email.posthoganalytics.feature_enabled", return_value=True)
+    @patch("posthog.email.requests.post", side_effect=requests.ReadTimeout("no response"))
+    def test_worker_retries_a_workflows_failure_on_the_http_schedule(
+        self,
+        _name: str,
+        retries: int,
+        expected_countdown: int | None,
+        expected_retrying: bool,
+        _mock_post: MagicMock,
+        _mock_feature_enabled: MagicMock,
+        mock_scoped_capture: MagicMock,
+        mock_apply_async: MagicMock,
+    ) -> None:
+        _send_email.push_request(id="worker-task", retries=retries, is_eager=False, called_directly=False)
+        try:
+            with (
+                patch("posthog.email.current_task", _send_email),
+                override_instance_config("EMAIL_HOST", "localhost"),
+                self.settings(
+                    CUSTOMER_IO_API_KEY="cio-key",
+                    POSTHOG_TRANSACTIONAL_EMAIL_API_KEY="phs_test",
+                    POSTHOG_HOST="https://us.posthog.com",
+                ),
+                patch.dict("posthog.email.WORKFLOWS_TEMPLATE_ID_MAP", {"password_reset": "tmpl-1"}),
+            ):
+                with self.assertRaises(Retry if expected_countdown else WorkflowsEmailError):
+                    _send_email.run(
+                        campaign_key="worker_campaign",
+                        to=[{"raw_email": "test@posthog.com", "recipient": "test@posthog.com"}],
+                        subject="Test subject",
+                        headers={},
+                        template_name="password_reset",
+                        use_http=True,
+                        properties={},
+                    )
+        finally:
+            _send_email.pop_request()
+
+        if expected_countdown is None:
+            mock_apply_async.assert_not_called()
+        else:
+            self.assertEqual(mock_apply_async.call_args.kwargs["countdown"], expected_countdown)
+        capture = mock_scoped_capture.return_value.__enter__.return_value
+        failed = [c.kwargs for c in capture.call_args_list if c.kwargs.get("event") == "transactional email failed"]
+        self.assertEqual([f["properties"]["retrying"] for f in failed], [expected_retrying])
+
     def test_sanitize_email_properties(self) -> None:
         # Test with various types of input including potential HTML injection
         properties = {