fix(metrics): honor send=False, fork-safe locks, flush on module shutdown

Address review of the metrics capture surface:

- _do_flush discards the window without transmitting when client.send is
  False, mirroring event capture (record locally, send nothing).
- _reinit_after_fork replaces the metrics locks (_metrics_lock, _lock,
  _flush_lock) and drops the inherited window/timer in the forked child, so
  a child can't deadlock on a lock held by a vanished parent thread.
- Module-level shutdown() delegates to Client.shutdown() so the final
  metrics window is flushed for apps using the global SDK lifecycle.

Generated-By: PostHog Code
Task-Id: 0d65d94c-c18a-4d84-93f6-18eb54876551

Author: DanielVisca

posthog/__init__.py

@@ -1132,8 +1132,7 @@ def shutdown() -> None:
     Category:
         Client management
     """
-    _proxy("flush")
-    _proxy("join")
+    _proxy("shutdown")
 
 
 def setup() -> Client:

posthog/client.py

@@ -1604,6 +1604,12 @@ def _reinit_after_fork(self):
         self._flag_definition_cache_provider_async_runner = None
         self._flag_definition_cache_provider_async_runner_lock = threading.Lock()
 
+        # Metrics locks may have been held by a parent thread at fork time; replace
+        # them (never acquire them) so the child can't deadlock on a vanished holder.
+        self._metrics_lock = threading.Lock()
+        if self._metrics is not None:
+            self._metrics._reinit_after_fork()
+
         # If using Redis cache, we must reinitialize to get a fresh connection (fork-safe).
         # If using Memory cache, we keep it as-is to benefit from the inherited warm cache.
         if isinstance(self.flag_cache, RedisFlagCache):

posthog/metrics_capture.py

@@ -376,14 +376,27 @@ def _capture(
             self._fold(state, float(value))
             self._arm_flush_timer()
 
+    def _reinit_after_fork(self) -> None:
+        # Runs in a forked child (via the client's os.register_at_fork hook) before
+        # user code. The inherited locks may be held by parent threads that do not
+        # exist in the child, so replace them without ever acquiring them.
+        self._lock = threading.Lock()
+        self._flush_lock = threading.Lock()
+        self._pid = os.getpid()
+        self._drop_inherited_window()
+
     def _reset_after_fork_locked(self) -> None:
-        # A forked child inherits the parent's window and a timer handle whose thread does
-        # not exist in the child — without this, the child never flushes (silent total
-        # loss) and would duplicate the parent's samples if it ever did. Drop both.
+        # PID-guard fallback for platforms without os.register_at_fork: a forked child
+        # inherits the parent's window and a timer handle whose thread does not exist
+        # in the child — without this, the child never flushes (silent total loss) and
+        # would duplicate the parent's samples if it ever did. Drop both.
         pid = os.getpid()
         if pid == self._pid:
             return
         self._pid = pid
+        self._drop_inherited_window()
+
+    def _drop_inherited_window(self) -> None:
         self._flush_timer = None
         self._series = {}
         self._series_cap_warned = False
@@ -447,6 +460,11 @@ def _do_flush(self) -> None:
             self._type_by_name = {}
             self._type_collision_warned = set()
 
+        # send=False mirrors event capture: recording succeeds locally, but
+        # nothing is transmitted — the flushed window is discarded.
+        if not getattr(self._client, "send", True):
+            return
+
         payload = self._build_payload(window)
         outcome = self._send(payload)
         if outcome == "retry-later":

posthog/test/test_client_fork.py

@@ -1,5 +1,6 @@
 import os
 import gc
+import signal
 import unittest
 import warnings
 import weakref
@@ -237,6 +238,46 @@ def child_probe():
         )
         self.assertEqual(result, "ok")
 
+    def test_register_at_fork_replaces_metrics_locks_in_child_process(self):
+        # Locks held at fork time are inherited locked, and their holders don't
+        # exist in the child — the metrics path must not deadlock on them.
+        client = Client(FAKE_TEST_API_KEY, send=False)
+        client.metrics.count("parent.metric")
+
+        locks = [
+            client._metrics_lock,
+            client.metrics._lock,
+            client.metrics._flush_lock,
+        ]
+
+        def child_probe():
+            # A deadlocked child gets killed by SIGALRM (a signaled exit fails
+            # the assertion below) instead of hanging the test forever.
+            signal.alarm(5)
+            try:
+                if not client._metrics_lock.acquire(blocking=False):
+                    return "inherited _metrics_lock still held"
+                client._metrics_lock.release()
+                client.metrics.count("child.metric")
+                client.metrics.flush()
+            finally:
+                signal.alarm(0)
+            return "ok"
+
+        for lock in locks:
+            lock.acquire()
+        try:
+            status, result = self._run_fork_probe(child_probe)
+        finally:
+            for lock in locks:
+                lock.release()
+            client.metrics.reset()
+
+        self.assertTrue(
+            os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, msg=result
+        )
+        self.assertEqual(result, "ok")
+
     def test_register_at_fork_reinitializes_poller_and_sessions_in_child_process(self):
         client = Client(
             FAKE_TEST_API_KEY,

posthog/test/test_metrics.py

@@ -5,6 +5,7 @@
 
 import pytest
 
+import posthog
 from posthog.client import Client
 from posthog.metrics_capture import DEFAULT_HISTOGRAM_BOUNDS
 from posthog.version import VERSION
@@ -241,6 +242,27 @@ def test_transient_failure_merges_window_back(self, client):
         (dp,) = metric["sum"]["dataPoints"]
         assert dp["asDouble"] == 7.0
 
+    @pytest.mark.parametrize(
+        "flush_via",
+        [
+            lambda m: m.flush(),  # manual flush
+            lambda m: m._timer_flush(),  # what the flush timer invokes
+        ],
+        ids=["manual", "timer"],
+    )
+    def test_send_false_aggregates_but_never_posts(self, flush_via):
+        # send=False documents "queueing succeeds but events are not sent"; metrics
+        # must mirror that: fold locally, never hit the transport.
+        c = Client(FAKE_API_KEY, host="https://us.example.com", send=False, thread=0)
+        c.metrics.count("jobs.processed")
+
+        session = mock_session()
+        with mock.patch("posthog.metrics_capture._get_session", return_value=session):
+            flush_via(c.metrics)
+        c.metrics.reset()
+
+        assert not session.post.called
+
     def test_shutdown_flushes_pending_metrics(self):
         c = Client(FAKE_API_KEY, host="https://us.example.com", sync_mode=True)
         c.metrics.count("m", 1)
@@ -254,6 +276,43 @@ def test_shutdown_flushes_pending_metrics(self):
         (metric,) = metrics_from(payload)
         assert metric["name"] == "m"
 
+    def test_module_shutdown_flushes_default_client_metrics(self):
+        # Module-level shutdown() must delegate to Client.shutdown(), or the default
+        # client from posthog.setup() loses its final metrics window.
+        saved = (
+            posthog.default_client,
+            posthog.api_key,
+            posthog.host,
+            posthog.sync_mode,
+        )
+        posthog.default_client = None
+        posthog.api_key = FAKE_API_KEY
+        posthog.host = "https://us.example.com"
+        posthog.sync_mode = True
+        try:
+            posthog.setup().metrics.count("m", 1)
+
+            session = mock_session()
+            with mock.patch(
+                "posthog.metrics_capture._get_session", return_value=session
+            ):
+                posthog.shutdown()
+
+            assert session.post.called
+            (metric,) = metrics_from(sent_payload(session))
+            assert metric["name"] == "m"
+
+            # The window was flushed and cleared, not left pending.
+            payload, _, _ = flush_and_capture(posthog.default_client)
+            assert payload is None
+        finally:
+            (
+                posthog.default_client,
+                posthog.api_key,
+                posthog.host,
+                posthog.sync_mode,
+            ) = saved
+
 
 class TestMetricsCrashSafety:
     # A telemetry SDK must never raise into the host application — these inputs all