diff --git a/HISTORY.md b/HISTORY.md index d858772..dc032fa 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -1,21 +1,26 @@ # Unreleased -### Upgrade note: new request headers and proxy allowlists + +### Upgrade note: new request headers + This release sends two request headers that earlier versions did not: `Authorization` (HTTP Basic, carrying your write key) and `X-Retry-Count` -(on retries only). If your traffic to Segment goes through a proxy, gateway +(sent on retries only). If traffic to Segment passes through a proxy, gateway or WAF that allowlists request headers, add both before upgrading or uploads will be rejected. -- Send the write key as an `Authorization: Basic` header. It is still included in the request body, so no server-side change is required. OAuth deployments continue to send `Authorization: Bearer` and are unaffected. -- Send `X-Retry-Count` on retries, so the server can distinguish a retry from a first attempt. Omitted on the first attempt. -- Unified retry handling: 429, 408, 410, 460 and 5xx (except 501 and 505) are retried. `Retry-After` is honoured on all of them, not just 429, which brings 529 in through the generic 5xx rule. 511 asks the client to re-authenticate, so it is retried only when an `oauth_manager` is configured and is dropped otherwise. -- `Retry-After` accepts numeric seconds and the RFC 7231 HTTP-date formats, capped at 300s. -- Rate-limited retries are bounded by elapsed time rather than counted against the retry limit, so a long `Retry-After` no longer exhausts the budget. -- New client options `max_total_backoff_duration` and `max_rate_limit_duration` (default 12 hours each) bound the two waits. -- Only 2xx responses count as a successful upload. A 3xx is now reported as a failed upload rather than silently treated as delivered. It is not retried: a redirect the HTTP client already declined to follow will not succeed on a retry. The Segment endpoint does not redirect, so this only affects custom `host` values. -- Backoff waits are interruptible, so `shutdown()` no longer blocks for the full delay. -- Retry timing uses a monotonic clock, so a system clock change cannot stretch or collapse a backoff. -- Fix a `queue.task_done()` leak that could leave `flush()` waiting forever when a batch was re-queued during shutdown. +### Retry handling + +- Uploads are retried on 408, 410, 429, 460, and 5xx except 501 and 505. 511 is retried only when an `oauth_manager` is configured, and dropped otherwise. +- A `Retry-After` header is honoured on any retryable response, not only 429. Numeric seconds and the RFC 7231 HTTP-date formats are both accepted, and the value is capped at 300 seconds. +- Responses carrying `Retry-After` are retried for up to `max_rate_limit_duration` and do not consume the retry count. A `Retry-After` that will not fit in what is left of the budget ends the episode rather than being shortened: retrying inside the window the server asked for sends a request it has already declined to serve, and the budget would be spent by then anyway. Other failures use exponential backoff from 500ms to a 60 second ceiling, limited by `max_retries` and by `max_total_backoff_duration` as an upper bound. +- New client options, both in seconds: `max_rate_limit_duration` (default 1800) and `max_total_backoff_duration` (default 43200). +- `flush()` and `shutdown()` are bounded by the same limits, and a pending retry does not delay shutdown. + +### Other changes + +- The write key is sent as an `Authorization: Basic` header. It remains in the request body, so no server-side change is required. Deployments using OAuth continue to send `Authorization: Bearer`. +- `X-Retry-Count` is sent on retries, allowing the server to distinguish a retry from a first attempt. It is omitted on the first attempt. +- Only 2xx responses count as a successful upload. A 3xx is reported as a failed upload rather than treated as delivered, and is not retried. The Segment endpoint does not redirect, so this affects only custom `host` values. # 2.3.6 / 2026-4-7 - Update and widen PyJWT version to address security issue diff --git a/segment/analytics/client.py b/segment/analytics/client.py index d5dc7e3..5063081 100644 --- a/segment/analytics/client.py +++ b/segment/analytics/client.py @@ -8,7 +8,12 @@ from dateutil.tz import tzutc -from segment.analytics.consumer import MAX_MSG_SIZE, Consumer +from segment.analytics.consumer import ( + DEFAULT_MAX_RATE_LIMIT_DURATION, + DEFAULT_MAX_TOTAL_BACKOFF_DURATION, + MAX_MSG_SIZE, + Consumer, +) from segment.analytics.oauth_manager import OauthManager from segment.analytics.request import DatetimeSerializer, post from segment.analytics.utils import clean, guess_timezone @@ -30,8 +35,8 @@ class DefaultConfig(object): gzip = False timeout = 15 max_retries = 10 - max_total_backoff_duration = 43200 - max_rate_limit_duration = 43200 + max_total_backoff_duration = DEFAULT_MAX_TOTAL_BACKOFF_DURATION + max_rate_limit_duration = DEFAULT_MAX_RATE_LIMIT_DURATION proxies = None thread = 1 upload_interval = 0.5 diff --git a/segment/analytics/consumer.py b/segment/analytics/consumer.py index deec68b..b12f0f4 100644 --- a/segment/analytics/consumer.py +++ b/segment/analytics/consumer.py @@ -24,7 +24,11 @@ class ShutdownInterrupted(Exception): # Default duration limits (12 hours in seconds) DEFAULT_MAX_TOTAL_BACKOFF_DURATION = 43200 -DEFAULT_MAX_RATE_LIMIT_DURATION = 43200 +# Rate-limited attempts are deliberately uncounted, so this duration is the only +# thing bounding them. Keep it several times MAX_RETRY_AFTER_SECONDS: at parity a +# single maximal Retry-After consumes the whole budget, leaving one attempt and no +# retry, and the cap stops binding because the remaining budget is always smaller. +DEFAULT_MAX_RATE_LIMIT_DURATION = 1800 class FatalError(Exception): @@ -166,7 +170,23 @@ def upload(self): # Still rate-limited; wait until the rate limit expires if self.rate_limited_until is not None: + remaining = self.max_rate_limit_duration - (now - self.rate_limit_start_time) wait_time = self.rate_limited_until - now + if wait_time > remaining: + # Shortening the wait to fit sends the next request inside the + # window the server asked us to wait out, which it has already said + # it will not serve, and the budget is spent by then so it would be + # the last attempt either way. + self.log.error( + "Rate limit budget (%ds) cannot accommodate the requested wait; dropping batch.", + self.max_rate_limit_duration, + ) + self.clear_rate_limit_state() + if self.on_error: + self.on_error(Exception("Rate limit duration exceeded, batch dropped"), batch) + for _ in batch: + self.queue.task_done() + return False if wait_time > 0: self.log.debug("Rate-limited. Waiting %.2fs before next upload attempt.", wait_time) if not self._wait(wait_time): @@ -201,11 +221,18 @@ def upload(self): self._requeue(batch) success = False else: + # The request completed carrying no rate-limit signal, so the episode + # is over. Nothing else clears the marker, so leaving it set drops the + # next batch to arrive after the budget elapses — for a rate limit that + # ended here, without ever sending it. + self.clear_rate_limit_state() self.log.error("error uploading: %s", e) success = False if self.on_error: self.on_error(e, batch) except Exception as e: + # Same reasoning as the non-rate-limited APIError branch above. + self.clear_rate_limit_state() self.log.error("error uploading: %s", e) success = False if self.on_error: diff --git a/segment/analytics/request.py b/segment/analytics/request.py index 8f3ee0e..09243ca 100644 --- a/segment/analytics/request.py +++ b/segment/analytics/request.py @@ -15,7 +15,10 @@ _session = sessions.Session() -# Maximum Retry-After delay to respect (5 minutes) +# A guard against an absurd header, not a second budget. Waiting less than the +# server asked for does not make the next attempt more likely to succeed, it just +# sends more requests at something already rate-limiting us; how long we keep +# trying is max_rate_limit_duration's job. MAX_RETRY_AFTER_SECONDS = 300 diff --git a/segment/analytics/test/test_consumer.py b/segment/analytics/test/test_consumer.py index 5bbc58b..e868b0b 100644 --- a/segment/analytics/test/test_consumer.py +++ b/segment/analytics/test/test_consumer.py @@ -10,8 +10,14 @@ except ImportError: from Queue import Queue -from segment.analytics.consumer import MAX_MSG_SIZE, Consumer, FatalError -from segment.analytics.request import APIError +from segment.analytics.client import Client +from segment.analytics.consumer import ( + DEFAULT_MAX_RATE_LIMIT_DURATION, + MAX_MSG_SIZE, + Consumer, + FatalError, +) +from segment.analytics.request import MAX_RETRY_AFTER_SECONDS, APIError class TestConsumer(unittest.TestCase): @@ -323,8 +329,8 @@ def mock_post_fn(*args, **kwargs): # rate_limited_until should be ~10 seconds in the future self.assertGreater(consumer.rate_limited_until, time.monotonic() + 5) - def test_retry_after_capped_at_300_seconds(self): - """Test that Retry-After delay is capped at 300 seconds when setting rate-limit state""" + def test_retry_after_capped_at_max_retry_after_seconds(self): + """Retry-After is clamped to MAX_RETRY_AFTER_SECONDS when setting rate-limit state.""" consumer = Consumer(None, "testsecret", retries=2) track = {"type": "track", "event": "python event", "userId": "userId"} @@ -340,11 +346,56 @@ def mock_post_fn(*args, **kwargs): with self.assertRaises(APIError): consumer.request([track]) - # rate_limited_until should be capped at ~300s from now (not 600s) + # rate_limited_until should be capped at ~300s from now, not 600s self.assertIsNotNone(consumer.rate_limited_until) self.assertLessEqual(consumer.rate_limited_until, now + 310) self.assertGreater(consumer.rate_limited_until, now + 290) + def test_a_wait_that_will_not_fit_the_budget_drops_instead_of_shortening(self): + """Never retry inside the window the server asked us to wait out. + + Shortening the wait to fit the budget sends the next request before the + server said it would serve one, and the budget is spent by then, so it + would be the last attempt regardless. Dropping here costs the same batch + and one fewer request against a server already rate-limiting us. + """ + consumer = Consumer(Queue(), "testsecret", max_rate_limit_duration=300) + consumer.queue.put({"type": "track", "event": "e", "userId": "u"}) + + errors = [] + consumer.on_error = lambda e, b: errors.append(e) + + # 30s of budget left against a Retry-After of 60: it cannot fit. + consumer.rate_limit_start_time = time.monotonic() - 270 + consumer.rate_limited_until = time.monotonic() + 60 + + waits = [] + posts = [] + with mock.patch.object(Consumer, "_wait", side_effect=lambda s: waits.append(s) or True): + with mock.patch("segment.analytics.consumer.post", side_effect=lambda *a, **k: posts.append(1)): + consumer.upload() + + self.assertEqual(waits, [], "should not have waited a shortened interval") + self.assertEqual(posts, [], "should not have sent a request inside the Retry-After window") + self.assertEqual(len(errors), 1, "the dropped batch must be reported") + + def test_a_wait_that_fits_the_budget_is_honoured_in_full(self): + """The counterpart: a wait that fits is taken as the server asked for it.""" + consumer = Consumer(Queue(), "testsecret", max_rate_limit_duration=300) + consumer.queue.put({"type": "track", "event": "e", "userId": "u"}) + + # 120s of budget left against a Retry-After of 60: it fits. + consumer.rate_limit_start_time = time.monotonic() - 180 + consumer.rate_limited_until = time.monotonic() + 60 + + waits = [] + with mock.patch.object(Consumer, "_wait", side_effect=lambda s: waits.append(s) or True): + with mock.patch("segment.analytics.consumer.post", return_value=None): + consumer.upload() + + self.assertTrue(waits, "expected the consumer to wait") + self.assertGreater(waits[0], 55, f"waited {waits[0]}s; the full Retry-After should be honoured") + def test_408_and_503_without_retry_after_use_backoff(self): """Test that 408 and 503 without Retry-After header use exponential backoff""" track = {"type": "track", "event": "python event", "userId": "userId"} @@ -1131,3 +1182,82 @@ def mock_post_fn(*args, **kwargs): self.assertEqual(error.status, 500) self.assertEqual(len(batch), 1) self.assertEqual(batch[0]["event"], "test event") + + def test_default_rate_limit_budget_exceeds_the_retry_after_cap(self): + """The budget must leave room for more than one maximal Retry-After. + + The elapsed check runs before the wait, so at parity a single capped + Retry-After spends the whole budget and the batch is dropped having been + attempted once, with no retry at all. + """ + self.assertGreater( + DEFAULT_MAX_RATE_LIMIT_DURATION, + MAX_RETRY_AFTER_SECONDS, + "a capped Retry-After would consume the entire rate-limit budget", + ) + self.assertGreaterEqual( + DEFAULT_MAX_RATE_LIMIT_DURATION // MAX_RETRY_AFTER_SECONDS, + 2, + "budget leaves room for fewer than two capped waits", + ) + + def test_client_defaults_to_the_documented_rate_limit_budget(self): + """Pins what a real caller gets, which is not the same as Consumer's default. + + Client passes its own DefaultConfig value into every Consumer it builds, so + asserting Consumer's default here would stay green while a drifted Client + default shipped. + """ + client = Client("testsecret", send=False) + try: + self.assertEqual( + client.consumers[0].max_rate_limit_duration, + DEFAULT_MAX_RATE_LIMIT_DURATION, + ) + finally: + client.shutdown() + + def test_non_rate_limited_failure_ends_the_episode(self): + """A request that completed without a rate-limit signal ends the episode. + + The marker outlives the batch that opened it, so leaving it set strands it: + upload() returns at the empty-batch guard before the budget block, and + nothing else clears it. + """ + q = Queue() + q.put({"event": "one"}) + consumer = Consumer(q, "testsecret", on_error=lambda e, b: None) + consumer.rate_limit_start_time = time.monotonic() + + with mock.patch( + "segment.analytics.consumer.post", + side_effect=APIError(400, "invalid", "Bad Request"), + ): + consumer.upload() + + self.assertIsNone( + consumer.rate_limit_start_time, + "a completed request carrying no rate-limit signal must end the episode", + ) + + def test_batch_after_an_ended_episode_is_still_sent(self): + """The symptom: a stranded marker drops a batch that was never rate-limited.""" + q = Queue() + consumer = Consumer(q, "testsecret", max_rate_limit_duration=1, on_error=lambda e, b: None) + consumer.rate_limit_start_time = time.monotonic() + + q.put({"event": "one"}) + with mock.patch( + "segment.analytics.consumer.post", + side_effect=APIError(400, "invalid", "Bad Request"), + ): + consumer.upload() + + time.sleep(1.1) # longer than this consumer's 1 second budget + + sent = [] + q.put({"event": "two"}) + with mock.patch("segment.analytics.consumer.post", side_effect=lambda *a, **k: sent.append(1)): + consumer.upload() + + self.assertEqual(len(sent), 1, "batch was dropped for a rate-limit episode that had already ended") diff --git a/segment/analytics/test/test_request.py b/segment/analytics/test/test_request.py index 2ac5f65..fb3dd24 100644 --- a/segment/analytics/test/test_request.py +++ b/segment/analytics/test/test_request.py @@ -156,7 +156,7 @@ def test_parse_retry_after_integer(self): self.assertEqual(result, 30) def test_parse_retry_after_capped(self): - """Test that Retry-After is capped at 300 seconds""" + """Retry-After is capped at MAX_RETRY_AFTER_SECONDS""" response = mock.Mock() response.headers = {"Retry-After": "600"} result = parse_retry_after(response)