Bound the rate-limit retry budget at 30 minutes

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

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

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:

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
 
 

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")

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)