From a8d2c9d2bdac81648c9b0bc75b39a9f23fb2f1a7 Mon Sep 17 00:00:00 2001 From: Andy Eschbacher Date: Thu, 27 Aug 2026 06:26:38 -0400 Subject: [PATCH] fix: various stravapipe issues --- .../cloudrun/postgres_writer_app.py | 36 ++++++++--- .../src/stravapipe/cloudrun/pubsub.py | 31 +++++----- packages/stravapipe/src/stravapipe/retry.py | 28 +++++++-- .../unit/adapters/strava/test_repositories.py | 9 ++- .../tests/unit/cloudrun/test_pubsub.py | 40 ++++++++++++ packages/stravapipe/tests/unit/test_retry.py | 61 ++++++++++++++++++- 6 files changed, 170 insertions(+), 35 deletions(-) diff --git a/packages/stravapipe/src/stravapipe/cloudrun/postgres_writer_app.py b/packages/stravapipe/src/stravapipe/cloudrun/postgres_writer_app.py index b23dc6fdf..bc2a0f660 100644 --- a/packages/stravapipe/src/stravapipe/cloudrun/postgres_writer_app.py +++ b/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 diff --git a/packages/stravapipe/src/stravapipe/cloudrun/pubsub.py b/packages/stravapipe/src/stravapipe/cloudrun/pubsub.py index 4b690b262..6abff8a21 100644 --- a/packages/stravapipe/src/stravapipe/cloudrun/pubsub.py +++ b/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", "") diff --git a/packages/stravapipe/src/stravapipe/retry.py b/packages/stravapipe/src/stravapipe/retry.py index 9a5ab1fc5..aab5c9b5c 100644 --- a/packages/stravapipe/src/stravapipe/retry.py +++ b/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, diff --git a/packages/stravapipe/tests/unit/adapters/strava/test_repositories.py b/packages/stravapipe/tests/unit/adapters/strava/test_repositories.py index 9428523fb..982df95e6 100644 --- a/packages/stravapipe/tests/unit/adapters/strava/test_repositories.py +++ b/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 diff --git a/packages/stravapipe/tests/unit/cloudrun/test_pubsub.py b/packages/stravapipe/tests/unit/cloudrun/test_pubsub.py index 0a8758784..6cbece0ca 100644 --- a/packages/stravapipe/tests/unit/cloudrun/test_pubsub.py +++ b/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 diff --git a/packages/stravapipe/tests/unit/test_retry.py b/packages/stravapipe/tests/unit/test_retry.py index de9a7e260..e36fc6b81 100644 --- a/packages/stravapipe/tests/unit/test_retry.py +++ b/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."""