From a85aa14543400978126f98fc309420458c75916f Mon Sep 17 00:00:00 2001 From: Harley Alexander <43975092+mayteio@users.noreply.github.com> Date: Thu, 29 Oct 2026 12:02:59 +0000 Subject: [PATCH 1/2] fix(email): stop falling back to customer.io for transactional emails A recipient on the Workflows suppression list is now dropped, like an unreachable address. Every other Workflows refusal retries on Workflows instead of sending through Customer.io. A retry stays unpinned when Workflows certainly queued nothing, so turning the flag off still moves it to Customer.io. Generated-By: PostHog Desktop Task-Id: 120fc7fd-3cc1-484a-99ee-da0e2036ca93 --- posthog/email.py | 42 +++++++++++-------------------- posthog/test/test_email.py | 51 ++++++++++++++++++++++++-------------- 2 files changed, 47 insertions(+), 46 deletions(-) diff --git a/posthog/email.py b/posthog/email.py index d37bfc6e483c..5685aa052d07 100644 --- a/posthog/email.py +++ b/posthog/email.py @@ -220,20 +220,14 @@ def is_email_available(with_absolute_urls: bool = False) -> bool: # 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( @@ -294,8 +288,8 @@ def __init__(self, message: str, *, code: str, status_code: int | None = None) - 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 +510,8 @@ def _send_one( 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 +534,11 @@ def _send_one( {"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) diff --git a/posthog/test/test_email.py b/posthog/test/test_email.py index 8b95080a0b30..27bc319b3924 100644 --- a/posthog/test/test_email.py +++ b/posthog/test/test_email.py @@ -505,22 +505,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 +549,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 +558,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,6 +695,7 @@ 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"})), @@ -702,6 +703,7 @@ def test_send_via_http_routes_through_workflows_when_flag_matches( ), ( "workflows_unreachable", + requests.ReadTimeout("no response"), True, "phs_test", requests.ConnectionError( @@ -716,19 +718,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 +751,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"), From a06f8664521c29457903f1537d58cf8b34598b85 Mon Sep 17 00:00:00 2001 From: Harley Alexander <43975092+mayteio@users.noreply.github.com> Date: Thu, 29 Oct 2026 12:03:01 +0000 Subject: [PATCH 2/2] fix(email): retry workflows sends for about 19 minutes Customer.io no longer covers a Workflows outage, so a worker now retries a failed HTTP send after 10s, 30s, 1m, 2m, 5m and 10m instead of 5s, 10s and 20s. The window stops at about 19 minutes because a login code or password reset that arrives later is of no use. SMTP sends keep their existing retries. Generated-By: PostHog Desktop Task-Id: 120fc7fd-3cc1-484a-99ee-da0e2036ca93 --- posthog/email.py | 34 +++++++++++++++++++---- posthog/test/test_email.py | 56 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 84 insertions(+), 6 deletions(-) diff --git a/posthog/email.py b/posthog/email.py index 5685aa052d07..6bdc0565fed0 100644 --- a/posthog/email.py +++ b/posthog/email.py @@ -108,7 +108,8 @@ def is_email_available(with_absolute_urls: bool = False) -> bool: "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, } @@ -282,6 +283,12 @@ def __init__(self, message: str, *, code: str, status_code: int | None = None) - 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. @@ -868,14 +875,20 @@ def sanitize_value(value: Any, *, key: Any = None) -> Any: 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( @@ -900,8 +913,17 @@ def _send_email_now( 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( diff --git a/posthog/test/test_email.py b/posthog/test/test_email.py index 27bc319b3924..70d8c5f07f47 100644 --- a/posthog/test/test_email.py +++ b/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, @@ -912,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 = {