fix: various stravapipe issues

packages/stravapipe/src/stravapipe/cloudrun/postgres_writer_app.py

@@ -333,12 +333,36 @@ async def _handle_create(
     # malformed payload will fail identically.
     activity = validate_or_422(StandardActivity, raw_activity, context="raw_activity")
 
+    # Decode the polyline BEFORE opening the transaction. It is pure CPU and
+    # dominates for long routes, so doing it inside the transaction held a Neon
+    # connection open across work that needs no database at all — and pooled
+    # connections are the scarce resource on this path (see the open task on
+    # bounding writer concurrency). Only insert_route / tag_activity_regions
+    # actually need the transaction.
+    #
+    # Cost of hoisting: on a redelivered CREATE that turns out to be
+    # ALREADY_EXISTS or RESURRECTION_BLOCKED we decode a polyline we then throw
+    # away. That is the minority path (Pub/Sub at-least-once redelivery) and it
+    # burns CPU we are not otherwise using, rather than a connection everything
+    # else is queued behind.
+    #
+    # Trace shape note: postgres.polyline.decode is now a SIBLING of
+    # postgres.insert rather than a child of it. The span still exists with the
+    # same name and attributes, but anything that asserted on its parent will
+    # see the new nesting.
+    geojson: str | None = None
+    if activity.map and activity.map.polyline:
+        with record_span(
+            tracer,
+            "postgres.polyline.decode",
+            {"desirelines.activity_id": activity_id},
+        ):
+            geojson = decode_polyline_to_geojson(activity.map.polyline)
+
     # Insert to PostgreSQL within transaction (no Strava API call needed).
     # The UoW emits postgres.session.acquire and postgres.commit sub-spans
     # internally; the call-site sub-spans below cover the work in between
-    # so a trace shows insert / polyline-decode / route-insert / commit
-    # latency separately. Polyline decode is pure CPU and dominates for
-    # long routes, which is why it gets its own span.
+    # so a trace shows insert / route-insert / commit latency separately.
     uow = SqlAlchemyUnitOfWork(session_factory, tracer=tracer)
     with (
         _pg_span(tracer, "postgres.insert", "INSERT", activity_id),
@@ -361,12 +385,6 @@ async def _handle_create(
             and activity.map
             and activity.map.polyline
         ):
-            with record_span(
-                tracer,
-                "postgres.polyline.decode",
-                {"desirelines.activity_id": activity_id},
-            ):
-                geojson = decode_polyline_to_geojson(activity.map.polyline)
             if geojson:
                 with _pg_span(
                     tracer, "postgres.activities.insert_route", "INSERT", activity_id

packages/stravapipe/src/stravapipe/cloudrun/pubsub.py

@@ -60,23 +60,21 @@ class CloudEventContext:
     delivery_attempt: int | None = None
 
 
-# Valid content types for CloudEvents
-# - application/json: Binary format (metadata in ce-* headers, data in body)
-# - application/cloudevents+json: Structured format (everything in JSON body)
+# Valid content types for CloudEvents.
 #
-# Only binary mode is actually implemented below — the parser reads ce-type /
-# ce-id / ce-source from headers, which a structured-mode request does not send.
-# A genuine structured-mode delivery therefore clears this gate and then 400s on
-# the missing headers. Eventarc sends binary mode, so nothing hits it today.
-# Either narrow this set to what is parsed or implement structured mode; note
-# there is no CloudEvents SDK dependency here by design, as the whole parser is
+# Binary mode only (metadata in ce-* headers, data in the body) — that is what
+# the hand-rolled parser below implements, and what Eventarc sends.
+#
+# ``application/cloudevents+json`` (structured mode: everything in the JSON body)
+# was previously accepted here even though nothing parses it. A structured-mode
+# delivery cleared this gate and then failed further down with a 400 about
+# missing ce-* headers, which reads as a malformed request rather than an
+# unsupported format. It is rejected up front with an accurate 415 instead.
+#
+# If structured mode is ever needed, implement it — do not just re-add the
+# string. There is deliberately no CloudEvents SDK dependency here; the parser is
 # hand-rolled from headers plus a base64 body.
-_VALID_CONTENT_TYPES = frozenset(
-    {
-        "application/json",
-        "application/cloudevents+json",
-    }
-)
+_VALID_CONTENT_TYPES = frozenset({"application/json"})
 
 
 async def parse_pubsub_cloudevent(
@@ -91,7 +89,8 @@ async def parse_pubsub_cloudevent(
         Tuple of (CloudEventContext, decoded message data, message attributes)
 
     Raises:
-        HTTPException: If parsing fails (400) or validation fails (422)
+        HTTPException: 415 if the content type is not binary-mode CloudEvents,
+            400 if parsing fails, 422 if validation fails.
     """
     # Validate content type
     content_type = request.headers.get("content-type", "")

packages/stravapipe/src/stravapipe/retry.py

@@ -31,6 +31,18 @@
 # ceiling a spec-valid header could pin a worker for hours. 900s = 15 min.
 MAX_RETRY_AFTER_SECONDS = 900
 
+# Spread added on top of a honored ``Retry-After`` so concurrent rate-limited
+# workers don't all wake on the same instant and re-trip the limit together.
+# Strava's limits reset on shared wall-clock boundaries, so every client tends
+# to be handed the *same* Retry-After — the exact lockstep the 5xx path already
+# jitters against.
+#
+# Deliberately additive, not full jitter: the 5xx path samples [0, nominal)
+# because it invented that delay itself, but Retry-After is the server telling
+# us the earliest acceptable retry. Sampling below it would retry early and earn
+# another 429, so the honored value is a floor and the jitter only ever delays.
+RATE_LIMIT_JITTER_SECONDS = 5.0
+
 
 def _parse_retry_after(
     header_value: str | None, *, default: int = DEFAULT_RETRY_AFTER_SECONDS
@@ -155,11 +167,19 @@ def wrapper(*args: Any, **kwargs: Any) -> Any:
                                     "attempts",
                                     retry_after=retry_after,
                                 ) from e
-                            # Cap the wait so a spec-valid but far-future
-                            # ``Retry-After`` can't block a worker indefinitely.
-                            sleep_seconds = min(retry_after, MAX_RETRY_AFTER_SECONDS)
+                            # Honor Retry-After as a floor, add a small spread
+                            # (see RATE_LIMIT_JITTER_SECONDS), then cap the whole
+                            # thing so a spec-valid but far-future Retry-After
+                            # can't block a worker indefinitely. Capping *last*
+                            # keeps MAX_RETRY_AFTER_SECONDS a true ceiling; the
+                            # jitter is simply absorbed in that pathological case.
+                            sleep_seconds = min(
+                                retry_after
+                                + random.uniform(0, RATE_LIMIT_JITTER_SECONDS),
+                                MAX_RETRY_AFTER_SECONDS,
+                            )
                             logger.warning(
-                                "Rate limited, waiting %s seconds (attempt %d/%d)",
+                                "Rate limited, waiting %.1f seconds (attempt %d/%d)",
                                 sleep_seconds,
                                 attempt + 1,
                                 max_attempts,

packages/stravapipe/tests/unit/adapters/strava/test_repositories.py

@@ -37,7 +37,7 @@
     StravaRateLimitError,
     StravaTokenError,
 )
-from stravapipe.retry import MAX_RETRY_AFTER_SECONDS
+from stravapipe.retry import MAX_RETRY_AFTER_SECONDS, RATE_LIMIT_JITTER_SECONDS
 
 
 @pytest.fixture
@@ -268,8 +268,11 @@ def test_get_activity_429_raises_rate_limit_error(self, api_client, api_config):
 
             # Every attempt hit the 429 before exhaustion.
             assert m.call_count == api_config.activity_retry_attempts
-            # Parsed Retry-After drove the backoff (not the 60s default)...
-            mock_sleep.assert_called_with(5)
+            # Parsed Retry-After drove the backoff (not the 60s default). It is
+            # a floor: RATE_LIMIT_JITTER_SECONDS is added on top to de-sync
+            # concurrent workers, so assert the window rather than an exact value.
+            (slept,) = mock_sleep.call_args.args
+            assert 5 <= slept <= 5 + RATE_LIMIT_JITTER_SECONDS
             # ...and is reported on the exception.
             assert exc_info.value.retry_after == 5
 

packages/stravapipe/tests/unit/cloudrun/test_pubsub.py

@@ -61,3 +61,43 @@ def test_delivery_attempt_is_none_when_absent(self, app):
         )
         assert response.status_code == 200
         assert response.json()["delivery_attempt"] is None
+
+
+class TestContentTypeGate:
+    """The accepted content types must match what the parser implements.
+
+    Only binary-mode CloudEvents are parsed (metadata in ce-* headers). Structured
+    mode — everything inside a JSON body under ``application/cloudevents+json`` —
+    used to be accepted here, cleared the gate, and then failed further down with
+    a 400 about missing ce-* headers. That reads as a malformed request rather
+    than an unsupported format. Audit 2026-08-06-stravapipe L1.
+    """
+
+    def test_binary_mode_is_accepted(self, app):
+        response = app.post(
+            "/",
+            headers=make_cloudevent_headers(),
+            json=make_pubsub_body(make_webhook_payload()),
+        )
+        assert response.status_code == 200
+
+    def test_structured_mode_is_rejected_as_unsupported_not_malformed(self, app):
+        headers = make_cloudevent_headers()
+        headers["content-type"] = "application/cloudevents+json"
+
+        response = app.post(
+            "/", headers=headers, json=make_pubsub_body(make_webhook_payload())
+        )
+
+        # 415, not 400: the format is unsupported, not the payload malformed.
+        assert response.status_code == 415
+        assert "application/cloudevents+json" in response.json()["detail"]
+
+    def test_charset_parameter_does_not_defeat_the_gate(self, app):
+        headers = make_cloudevent_headers()
+        headers["content-type"] = "application/json; charset=utf-8"
+
+        response = app.post(
+            "/", headers=headers, json=make_pubsub_body(make_webhook_payload())
+        )
+        assert response.status_code == 200

packages/stravapipe/tests/unit/test_retry.py

@@ -13,7 +13,11 @@
 import requests
 
 from stravapipe.exceptions import StravaRateLimitError
-from stravapipe.retry import _parse_retry_after, retry_on_failure
+from stravapipe.retry import (
+    RATE_LIMIT_JITTER_SECONDS,
+    _parse_retry_after,
+    retry_on_failure,
+)
 
 
 class TestRetryOnFailure:
@@ -116,7 +120,11 @@ def rate_limited_func():
             result = rate_limited_func()
             assert result == "success"
             assert call_count == 2
-            mock_sleep.assert_called_once_with(1)
+            # Retry-After is a floor, not an exact sleep: RATE_LIMIT_JITTER_SECONDS
+            # is added on top so concurrent rate-limited workers don't wake in
+            # lockstep. Never sleep less than the server asked for.
+            (slept,) = mock_sleep.call_args.args
+            assert 1 <= slept <= 1 + RATE_LIMIT_JITTER_SECONDS
 
     def test_rate_limit_429_exceeds_max_attempts(self):
         """Test rate limiting that exceeds max attempts."""
@@ -156,7 +164,54 @@ def rate_limited_no_header():
         with patch("time.sleep") as mock_sleep:
             result = rate_limited_no_header()
             assert result == "success"
-            mock_sleep.assert_called_once_with(60)  # Default fallback
+            # 60s default fallback, plus the anti-lockstep jitter.
+            (slept,) = mock_sleep.call_args.args
+            assert 60 <= slept <= 60 + RATE_LIMIT_JITTER_SECONDS
+
+    def test_rate_limit_jitter_desynchronises_and_never_undercuts_retry_after(self):
+        """429 backoff spreads wakeups without ever retrying early.
+
+        The 5xx path uses AWS full jitter (sample from ``[0, nominal)``) because
+        it invents its own delay. ``Retry-After`` is different: it is the server
+        stating the earliest acceptable retry, so sampling below it would retry
+        early and earn another 429. The jitter is therefore additive — a floor of
+        ``retry_after`` with a bounded spread above it.
+
+        Regression for audit 2026-07-30-stravapipe L1: this path previously slept
+        exactly ``Retry-After``, so every rate-limited worker woke on the same
+        instant and re-tripped the limit together.
+        """
+        retry_after = 30
+        observed: list[float] = []
+
+        for _ in range(40):
+            call_count = 0
+
+            @retry_on_failure(max_attempts=2, backoff_seconds=0.01)
+            def rate_limited():
+                nonlocal call_count
+                call_count += 1
+                if call_count < 2:
+                    response = Mock()
+                    response.status_code = 429
+                    response.headers = {"Retry-After": str(retry_after)}
+                    error = requests.exceptions.HTTPError("Rate limited")
+                    error.response = response
+                    raise error
+                return "success"
+
+            with patch("time.sleep") as mock_sleep:
+                assert rate_limited() == "success"
+                (slept,) = mock_sleep.call_args.args
+                observed.append(slept)
+
+        # Never earlier than the server permitted, never unboundedly later.
+        assert all(
+            retry_after <= s <= retry_after + RATE_LIMIT_JITTER_SECONDS
+            for s in observed
+        )
+        # Actually spread — a fixed sleep would collapse to a single value.
+        assert len(set(observed)) > 1
 
     def test_exponential_backoff(self):
         """Nominal backoff progression with jitter pinned to the upper bound."""