fix: refuse email sends after shutdown and drain async sends first

client.emails now asks the parent client to admit each send. After shutdown, a kept or new client.emails raises EmailError with code client_closed and makes no request. On AsyncClient, an admitted send is an in-flight operation, so shutdown waits for it, retries included, before it closes the email transport.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Generated-By: PostHog Desktop
Task-Id: c62b30cd-eb2e-496c-a73d-2a6e7a52a62e

Author: mayteio

posthog/async_client.py

@@ -10,7 +10,7 @@
 import warnings
 import weakref
 from datetime import datetime, timezone
-from typing import Any, Dict, Mapping, Optional, Union
+from typing import Any, Callable, Dict, Mapping, Optional, Union
 from uuid import UUID, uuid4
 
 from typing_extensions import Unpack
@@ -271,10 +271,21 @@ def emails(self) -> AsyncEmails:
         """
         if self._emails is None:
             self._emails = _client_emails(
-                AsyncEmails, self.secret_key, self.host, lambda: self.disabled
+                AsyncEmails,
+                self.secret_key,
+                self.host,
+                lambda: self.disabled,
+                self._admit_email_send,
             )
         return self._emails
 
+    def _admit_email_send(self) -> Optional[Callable[[], None]]:
+        # Shutdown waits for in-flight operations before it closes the email transport.
+        if not self._accepting or self._closed:
+            return None
+        completion = self._start_inflight_operation()
+        return lambda: self._finish_inflight_operation(completion)
+
     async def __aenter__(self) -> AsyncClient:
         self._ensure_workers_started()
         return self

posthog/client.py

@@ -52,7 +52,7 @@
     set_context_session as _context_set_context_session,
     tag as _context_tag,
 )
-from posthog.emails import Emails, _client_emails
+from posthog.emails import Emails, _client_emails, _no_op
 from posthog.exception_capture import ExceptionCapture
 from posthog._logging import _configure_posthog_logging
 from posthog.exception_utils import (
@@ -1155,7 +1155,11 @@ def emails(self) -> Emails:
         """
         if self._emails is None:
             self._emails = _client_emails(
-                Emails, self.secret_key, self.host, lambda: self.disabled
+                Emails,
+                self.secret_key,
+                self.host,
+                lambda: self.disabled,
+                lambda: None if self._shutdown_requested else _no_op,
             )
         return self._emails
 

posthog/emails.py

@@ -281,6 +281,10 @@ def _retry_delay(error: EmailError, attempt: int, max_retries: int) -> Optional[
     return max(backoff, error.retry_after or 0.0)
 
 
+def _no_op() -> None:
+    pass
+
+
 def _log_warnings(result: SendEmailResult) -> None:
     for warning in result.warnings:
         log.warning("PostHog email warning %s: %s", warning.code, warning.message)
@@ -323,11 +327,21 @@ def __init__(
         self.max_retries = max_retries
         self.disabled = disabled
         self._parent_disabled: Callable[[], bool] = lambda: False
+        self._admit: Callable[[], Optional[Callable[[], None]]] = lambda: _no_op
         self._host_error: Optional[str] = None
 
     def _is_disabled(self) -> bool:
         return self.disabled or self._parent_disabled()
 
+    def _start_send(self) -> Callable[[], None]:
+        finish = self._admit()
+        if finish is None:
+            raise _sdk_error(
+                "client_closed",
+                "The PostHog client is shut down, so no email was sent.",
+            )
+        return finish
+
     def _secret_key(self) -> str:
         if self._host_error:
             raise _sdk_error("app_host_required", self._host_error)
@@ -423,6 +437,13 @@ def send(
         if self._is_disabled():
             log.debug("PostHog client is disabled, so no email was sent.")
             return SendEmailResult(id=None)
+        finish = self._start_send()
+        try:
+            return self._deliver(prepared)
+        finally:
+            finish()
+
+    def _deliver(self, prepared: _PreparedSend) -> SendEmailResult:
         auth = _BearerAuth(self._secret_key(), prepared.idempotency_key)
         session = self._get_session()
         url = self.host + prepared.path
@@ -539,6 +560,13 @@ async def send(
         if self._is_disabled():
             log.debug("PostHog client is disabled, so no email was sent.")
             return SendEmailResult(id=None)
+        finish = self._start_send()
+        try:
+            return await self._deliver(prepared)
+        finally:
+            finish()
+
+    async def _deliver(self, prepared: _PreparedSend) -> SendEmailResult:
         headers = _headers(self._secret_key(), prepared.idempotency_key)
         httpx, client = self._get_client()
         url = self.host + prepared.path
@@ -610,6 +638,7 @@ def _client_emails(
     secret_key: Optional[str],
     client_host: Optional[str],
     is_disabled: Callable[[], bool],
+    admit: Callable[[], Optional[Callable[[], None]]],
 ) -> _EmailsT:
     app_host = determine_app_host(client_host)
     is_cloud = app_host in (US_APP_ENDPOINT, EU_APP_ENDPOINT)
@@ -621,4 +650,5 @@ def _client_emails(
             "Create posthog.emails.Emails(secret_key, host=<your PostHog app host>) instead."
         )
     emails._parent_disabled = is_disabled
+    emails._admit = admit
     return emails

posthog/test/test_emails.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
 from datetime import datetime, timezone
@@ -305,6 +306,21 @@ def test_a_disabled_client_sends_nothing():
     assert adapter.requests == []
 
 
+def test_a_shut_down_client_sends_no_email():
+    client = Posthog("phc_test", secret_key=SECRET_KEY, sync_mode=True)
+    emails = client.emails
+
+    client.shutdown()
+
+    with mock.patch.object(requests.Session, "request") as request:
+        for handle in (emails, client.emails):
+            with pytest.raises(EmailError) as raised:
+                send(handle)
+            assert raised.value.code == "client_closed"
+    request.assert_not_called()
+    assert emails._session is None
+
+
 @parameterized.expand(
     [
         ("several_recipients", {"to": ["a@example.com", "b@example.com"]}, TypeError),
@@ -361,3 +377,40 @@ async def test_async_disabled_client_sends_nothing():
 
     assert result.id is None
     assert emails._client is None
+
+
+@pytest.mark.asyncio
+async def test_async_shutdown_waits_for_a_send_in_flight_and_then_refuses_sends():
+    started, release = asyncio.Event(), asyncio.Event()
+    replies = [
+        httpx.Response(503, json={"code": "delivery_unavailable"}),
+        httpx.Response(202, json={"message_id": "msg_1"}),
+    ]
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        started.set()
+        await release.wait()
+        return replies.pop(0)
+
+    client = AsyncPosthog("phc_test", secret_key=SECRET_KEY)
+    emails = client.emails
+    emails._client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+
+    with mock.patch("posthog.emails._retry_delay", return_value=0.0):
+        sending = asyncio.create_task(
+            emails.send(to="user@example.com", template_id=TEMPLATE_ID)
+        )
+        await started.wait()
+        shutting_down = asyncio.create_task(client.shutdown())
+        for _ in range(5):
+            await asyncio.sleep(0)
+        assert not shutting_down.done()
+        release.set()
+
+        assert (await sending).id == "msg_1"
+        await shutting_down
+
+    assert emails._client is None
+    with pytest.raises(EmailError) as raised:
+        await emails.send(to="user@example.com", template_id=TEMPLATE_ID)
+    assert raised.value.code == "client_closed"