fix(metrics): harden attribute snapshots, retry backoff, and config validation

.sampo/changesets/metrics-review-followups.md

@@ -0,0 +1,9 @@
+---
+pypi/posthog: patch
+---
+
+Harden the alpha `posthog.metrics` client based on review follow-ups.
+
+- Metric attributes are now deep-snapshotted at capture time, so mutating a nested list/dict value after `count()`/`gauge()`/`histogram()` can no longer rewrite an already-recorded series' attributes on the wire.
+- Failed metric flushes now retry with exponential backoff (2x per consecutive failure, capped at 64x the flush interval, matching the shared JS logs policy) instead of the fixed cadence, and the buffered window is dropped loudly after 8 consecutive failed flushes — previously documented as 3 but effectively 4.
+- Invalid `metrics` client config (non-dict config or `resource_attributes`, non-numeric `flush_interval`, non-integer `max_series_per_flush`, non-callable `before_send`) now degrades to defaults with a warning instead of raising from the first `client.metrics.count()` call, matching the client's no-throw contract.

posthog/client.py

@@ -1747,7 +1747,18 @@ def metrics(self) -> PostHogMetrics:
         if self._metrics is None:
             with self._metrics_lock:
                 if self._metrics is None:
-                    self._metrics = PostHogMetrics(self, self._metrics_config)
+                    # Same no-throw semantics as the rest of the public client surface:
+                    # a bad metrics config degrades to defaults instead of raising from
+                    # the first chained metrics.count() call (raise only in debug mode).
+                    try:
+                        self._metrics = PostHogMetrics(self, self._metrics_config)
+                    except Exception as e:
+                        if self.debug:
+                            raise e
+                        self.log.exception(
+                            f"Error initializing metrics, using default configuration: {e}"
+                        )
+                        self._metrics = PostHogMetrics(self, None)
         return self._metrics
 
     def flush(self, timeout_seconds: Optional[float] = 10) -> None:

posthog/metrics_capture.py

@@ -14,10 +14,13 @@
 
 Delivery is at-least-once: a request that succeeds server-side but fails
 client-side (e.g. a read timeout) is retried with the next window, which can
-double-count that window's deltas. Windows are dropped loudly after
-``_MAX_CONSECUTIVE_SEND_FAILURES`` failed flushes.
+double-count that window's deltas. Failed flushes retry with exponential
+backoff capped at ``_MAX_RETRY_BACKOFF_MULTIPLIER`` times the flush interval
+(the policy the shared JS logs implementation uses); the window is dropped
+loudly once ``_MAX_CONSECUTIVE_SEND_FAILURES`` consecutive flushes have failed.
 """
 
+import copy
 import gzip
 import json
 import logging
@@ -61,8 +64,13 @@
 _OTLP_TEMPORALITY_DELTA = 1
 _VALID_METRIC_TYPES = ("count", "gauge", "histogram")
 # Consecutive failed flushes before the buffered window is dropped (loudly) — bounds
-# memory and payload growth against a permanently unreachable endpoint.
-_MAX_CONSECUTIVE_SEND_FAILURES = 3
+# memory and payload growth against a permanently unreachable endpoint. The series
+# cap already bounds the buffered window, and backoff spaces the attempts out, so
+# the budget covers a real outage (~21 min at the default 10s interval).
+_MAX_CONSECUTIVE_SEND_FAILURES = 8
+# Retry delays grow 2x per consecutive failure, capped at this multiple of the
+# flush interval — the same ceiling the shared JS logs implementation uses.
+_MAX_RETRY_BACKOFF_MULTIPLIER = 64
 _DEFAULT_FLUSH_INTERVAL_SECONDS = 10.0
 _DEFAULT_MAX_SERIES_PER_FLUSH = 1000
 _SCOPE_NAME = "posthog-python"
@@ -157,9 +165,18 @@ def __init__(
         self.name = name
         self.type = metric_type
         self.unit = unit
-        # Snapshot: the series key was computed from these values, so a caller
-        # mutating the dict after capture must not change the stored series.
-        self.attributes = dict(attributes) if attributes else None
+        # Deep snapshot: the series key was computed from these values, so a caller
+        # mutating the dict — or a nested list/dict value — after capture must not
+        # change the stored series.
+        if attributes:
+            try:
+                self.attributes: Optional[dict] = copy.deepcopy(attributes)
+            except Exception:
+                # Un-copyable exotic values: a shallow snapshot still isolates the
+                # top-level dict, and the encoder stringifies whatever remains.
+                self.attributes = dict(attributes)
+        else:
+            self.attributes = None
         self.window_start_ms = int(time.time() * 1000)
         self.total: Optional[float] = None
         self.last: Optional[float] = None
@@ -181,8 +198,27 @@ class PostHogMetrics:
 
     def __init__(self, client, config: Optional[dict] = None):
         self._client = client
-        config = config or {}
-        resource_attributes = config.get("resource_attributes") or {}
+        # client.metrics sits outside the client's no-throw guards, so invalid nested
+        # config must degrade to defaults (with a warning) instead of raising into
+        # the host application from the first metrics.count() call. The Any-typed
+        # local keeps the runtime defense visible to mypy despite the annotation.
+        raw_config: Any = config
+        if not isinstance(raw_config, dict):
+            if raw_config is not None:
+                log.warning(
+                    "Ignoring metrics config: expected a dict, got %s",
+                    type(raw_config).__name__,
+                )
+            raw_config = {}
+        config = raw_config
+        resource_attributes = config.get("resource_attributes")
+        if not isinstance(resource_attributes, dict):
+            if resource_attributes is not None:
+                log.warning(
+                    "Ignoring metrics resource_attributes: expected a dict, got %s",
+                    type(resource_attributes).__name__,
+                )
+            resource_attributes = {}
         self._service_name: Optional[str] = resource_attributes.get(
             "service.name"
         ) or config.get("service_name")
@@ -193,13 +229,35 @@ def __init__(self, client, config: Optional[dict] = None):
             "deployment.environment"
         ) or config.get("environment")
         self._resource_attributes: dict = resource_attributes
-        self._flush_interval: float = config.get(
-            "flush_interval", _DEFAULT_FLUSH_INTERVAL_SECONDS
-        )
-        self._max_series_per_flush: int = config.get(
-            "max_series_per_flush", _DEFAULT_MAX_SERIES_PER_FLUSH
-        )
-        self._before_send: Optional[Callable] = config.get("before_send")
+        flush_interval = config.get("flush_interval", _DEFAULT_FLUSH_INTERVAL_SECONDS)
+        if (
+            not isinstance(flush_interval, (int, float))
+            or isinstance(flush_interval, bool)
+            or not flush_interval > 0
+        ):
+            log.warning(
+                "Ignoring metrics flush_interval %r: expected a positive number of seconds",
+                flush_interval,
+            )
+            flush_interval = _DEFAULT_FLUSH_INTERVAL_SECONDS
+        self._flush_interval: float = float(flush_interval)
+        max_series = config.get("max_series_per_flush", _DEFAULT_MAX_SERIES_PER_FLUSH)
+        if (
+            not isinstance(max_series, int)
+            or isinstance(max_series, bool)
+            or max_series <= 0
+        ):
+            log.warning(
+                "Ignoring metrics max_series_per_flush %r: expected a positive integer",
+                max_series,
+            )
+            max_series = _DEFAULT_MAX_SERIES_PER_FLUSH
+        self._max_series_per_flush: int = max_series
+        before_send = config.get("before_send")
+        if before_send is not None and not callable(before_send):
+            log.warning("Ignoring metrics before_send: expected a callable")
+            before_send = None
+        self._before_send: Optional[Callable] = before_send
 
         self._lock = threading.Lock()
         self._pid = os.getpid()
@@ -427,10 +485,12 @@ def _fold(self, state: _SeriesState, value: float) -> None:
                 _bucket_index_for(value, DEFAULT_HISTOGRAM_BOUNDS)
             ] += 1
 
-    def _arm_flush_timer(self) -> None:
+    def _arm_flush_timer(self, delay: Optional[float] = None) -> None:
         if self._flush_timer is not None:
             return
-        timer = threading.Timer(self._flush_interval, self._timer_flush)
+        timer = threading.Timer(
+            delay if delay is not None else self._flush_interval, self._timer_flush
+        )
         timer.daemon = True
         self._flush_timer = timer
         timer.start()
@@ -470,7 +530,7 @@ def _do_flush(self) -> None:
         if outcome == "retry-later":
             with self._lock:
                 self._consecutive_send_failures += 1
-                if self._consecutive_send_failures > _MAX_CONSECUTIVE_SEND_FAILURES:
+                if self._consecutive_send_failures >= _MAX_CONSECUTIVE_SEND_FAILURES:
                     # A persistently unreachable endpoint must not buffer forever: drop the
                     # window loudly instead of growing until a too-large drop loses more.
                     log.error(
@@ -482,15 +542,20 @@ def _do_flush(self) -> None:
                     self._consecutive_send_failures = 0
                     return
                 # Transient failure: merge the unsent window back so the data rides the
-                # next flush instead of being lost — and re-arm the timer, since with no
-                # new captures nothing else would schedule that flush.
+                # next flush instead of being lost — and re-arm the timer with capped
+                # exponential backoff, so a real outage isn't hammered at the base
+                # cadence. New captures see the armed timer and don't shorten it.
+                delay = self._flush_interval * min(
+                    2**self._consecutive_send_failures, _MAX_RETRY_BACKOFF_MULTIPLIER
+                )
                 log.warning(
-                    "Metrics flush failed (attempt %s of %s); will retry with the next window",
+                    "Metrics flush failed (attempt %s of %s); retrying in %.0fs",
                     self._consecutive_send_failures,
-                    _MAX_CONSECUTIVE_SEND_FAILURES + 1,
+                    _MAX_CONSECUTIVE_SEND_FAILURES,
+                    delay,
                 )
                 self._merge_window_back(window)
-                self._arm_flush_timer()
+                self._arm_flush_timer(delay)
         elif outcome == "too-large":
             log.warning(
                 "Metrics batch exceeded the server size limit and was dropped. "

posthog/test/test_metrics.py

@@ -7,7 +7,12 @@
 
 import posthog
 from posthog.client import Client
-from posthog.metrics_capture import DEFAULT_HISTOGRAM_BOUNDS
+from posthog.metrics_capture import (
+    _DEFAULT_FLUSH_INTERVAL_SECONDS,
+    _DEFAULT_MAX_SERIES_PER_FLUSH,
+    _MAX_CONSECUTIVE_SEND_FAILURES,
+    DEFAULT_HISTOGRAM_BOUNDS,
+)
 from posthog.version import VERSION
 
 FAKE_API_KEY = "phc_test_key"
@@ -343,6 +348,66 @@ def test_non_string_attribute_keys_stringify_on_the_wire(self, client):
         assert keys == {"a", "2"}
         assert all(isinstance(attr["key"], str) for attr in dp["attributes"])
 
+    def test_nested_attribute_values_snapshot_at_capture(self, client):
+        # The series key is computed at capture time; a caller mutating a nested
+        # value afterwards must not rewrite the stored series' attributes, or the
+        # wire payload diverges from the identity the series was keyed under.
+        tags = ["a"]
+        client.metrics.count("m", 1, attributes={"tags": tags})
+        tags.append("b")
+        client.metrics.count("m", 1, attributes={"tags": tags})
+
+        payload, _, _ = flush_and_capture(client)
+
+        (metric,) = metrics_from(payload)
+        points = metric["sum"]["dataPoints"]
+        assert len(points) == 2
+        wire_tags = sorted(
+            [
+                v["stringValue"]
+                for v in dp["attributes"][0]["value"]["arrayValue"]["values"]
+            ]
+            for dp in points
+        )
+        assert wire_tags == [["a"], ["a", "b"]]
+
+    @pytest.mark.parametrize(
+        "bad_config",
+        [
+            "not-a-dict",
+            {"resource_attributes": "bad"},
+            {"flush_interval": "10"},
+            {"max_series_per_flush": "many"},
+            {"before_send": "not-callable"},
+        ],
+        ids=[
+            "config-not-dict",
+            "resource-attributes-not-dict",
+            "flush-interval-not-number",
+            "series-cap-not-int",
+            "before-send-not-callable",
+        ],
+    )
+    def test_hostile_metrics_config_does_not_raise_and_still_records(self, bad_config):
+        # client.metrics is reached before the guarded capture path, so bad nested
+        # config must fall back to defaults instead of raising into the host app.
+        c = Client(
+            FAKE_API_KEY,
+            host="https://us.example.com",
+            sync_mode=True,
+            metrics=bad_config,
+        )
+        c.metrics.count("m", 1)  # the complete public call must not raise
+
+        assert c.metrics._flush_interval == _DEFAULT_FLUSH_INTERVAL_SECONDS
+        assert c.metrics._max_series_per_flush == _DEFAULT_MAX_SERIES_PER_FLUSH
+
+        payload, _, _ = flush_and_capture(c)
+        c.metrics.reset()
+
+        (metric,) = metrics_from(payload)
+        assert metric["name"] == "m"
+
     def test_list_attribute_records_as_array_value(self, client):
         client.metrics.count("arr", 1, attributes={"tags": ["a", "b"]})
 
@@ -475,11 +540,55 @@ def test_window_dropped_after_consecutive_failures(self, client):
         with mock.patch(
             "posthog.metrics_capture._get_session", return_value=mock_session(503)
         ):
-            for _ in range(4):
+            for _ in range(_MAX_CONSECUTIVE_SEND_FAILURES):
                 client.metrics.flush()
 
         payload, _, _ = flush_and_capture(client)
 
         assert (
             payload is None
         )  # budget exhausted → window dropped, nothing left to send
+
+    def test_window_survives_failures_until_the_drop_limit(self, client):
+        # One failure short of the budget the window must still deliver in full —
+        # dropping earlier than documented silently loses data during outages.
+        client.metrics.count("m", 3)
+        with mock.patch(
+            "posthog.metrics_capture._get_session", return_value=mock_session(503)
+        ):
+            for _ in range(_MAX_CONSECUTIVE_SEND_FAILURES - 1):
+                client.metrics.flush()
+
+        payload, _, _ = flush_and_capture(client)
+
+        (metric,) = metrics_from(payload)
+        (dp,) = metric["sum"]["dataPoints"]
+        assert dp["asDouble"] == 3.0
+
+    def test_failed_flushes_back_off_exponentially_capped(self):
+        # Retrying a down endpoint at the fixed cadence hammers it for the whole
+        # outage; retry delays must grow exponentially and cap at 64x the base
+        # interval (matching the shared JS logs policy), then reset on success.
+        c = Client(
+            FAKE_API_KEY,
+            host="https://us.example.com",
+            sync_mode=True,
+            metrics={"flush_interval": 1.0},
+        )
+        c.metrics.count("m", 1)
+
+        intervals = []
+        with mock.patch(
+            "posthog.metrics_capture._get_session", return_value=mock_session(503)
+        ):
+            for _ in range(_MAX_CONSECUTIVE_SEND_FAILURES - 1):
+                c.metrics._timer_flush()  # what the flush timer thread invokes
+                intervals.append(c.metrics._flush_timer.interval)
+        assert intervals == [2.0, 4.0, 8.0, 16.0, 32.0, 64.0, 64.0]
+
+        # A successful flush resets the backoff: the next capture arms the base interval.
+        flush_and_capture(c)
+        c.metrics.reset()
+        c.metrics.count("m", 1)
+        assert c.metrics._flush_timer.interval == 1.0
+        c.metrics.reset()