Handle Retry-After on every retryable status, including 529

Route any retryable response carrying a valid Retry-After header through the rate-limit path (no retry-budget cost) instead of special-casing 429. Retryable statuses without Retry-After continue to use counted exponential backoff. Adds 529 to the retryable set and covers both paths with tests. Matches the behaviour already shipped in analytics-java 3.5.5 and the generic-retry-after conformance suite in sdk-e2e-tests.
MichaelGHSeg · Sep 23, 2026 · Feb 12, 2026 · Feb 12, 2026 · Feb 12, 2026 · Feb 18, 2026
commit 9818b3418e77f5f0fdde736c369c486931f855c4
diff --git a/segment/analytics/consumer.py b/segment/analytics/consumer.py
@@ -133,19 +133,18 @@ def upload(self):
             self.clear_rate_limit_state()
             success = True
         except APIError as e:
-            if e.status == 429 and self.rate_limited_until is not None:
-                # 429: rate-limit state already set by request(). Re-queue batch.
-                self.log.debug('429 received. Re-queuing batch and halting upload iteration.')
+            if self.rate_limited_until is not None:
+                self.log.debug('Rate-limited (status %d). Re-queuing batch and halting upload iteration.', e.status)
                 dropped = []
                 for item in batch:
                     try:
                         self.queue.put(item, block=False)
                     except Exception:
                         dropped.append(item)
                 if dropped:
-                    self.log.error('Queue full during 429 re-queue. Dropping %d item(s).', len(dropped))
+                    self.log.error('Queue full during rate-limit re-queue. Dropping %d item(s).', len(dropped))
                     if self.on_error:
-                        self.on_error(Exception('Queue full, items dropped during 429 re-queue'), dropped)
+                        self.on_error(Exception('Queue full, items dropped during rate-limit re-queue'), dropped)
                 success = False
             else:
                 self.log.error('error uploading: %s', e)
@@ -279,21 +278,19 @@ def apply_backoff(e, label):
                 raise
 
             except APIError as e:
-                # 429 with valid Retry-After > 0: block the pipeline and let
-                # upload() re-queue the batch. Retry-After: 0 or missing falls
-                # through to counted backoff to avoid a tight re-queue loop.
-                if e.status == 429:
-                    retry_after = parse_retry_after(e.response) if e.response is not None else None
-                    if retry_after is not None and retry_after > 0:
-                        self.set_rate_limit_state(e.response)
-                        raise
-
                 if not is_retryable_status(e.status):
                     self.log.error(
                         f"Non-retryable error {e.status} after {total_attempts} attempts: {e}"
                     )
                     raise
 
+                # Any retryable status with valid Retry-After > 0: block pipeline, re-queue
+                retry_after = parse_retry_after(e.response) if e.response is not None else None
+                if retry_after is not None and retry_after > 0:
+                    self.set_rate_limit_state(e.response)
+                    raise
+
+                # No Retry-After: counted backoff
                 delay = apply_backoff(e, f"Retry attempt (status {e.status})")
                 time.sleep(delay)
 

diff --git a/segment/analytics/request.py b/segment/analytics/request.py
@@ -1,9 +1,11 @@
 from datetime import date, datetime
+from email.utils import parsedate_to_datetime
 from io import BytesIO
 from gzip import GzipFile
 import logging
 import json
 import base64
+import time as _time
 from dateutil.tz import tzutc
 
 from requests import sessions
@@ -31,8 +33,16 @@ def parse_retry_after(response):
         delay = int(retry_after)
         return min(max(delay, 0), MAX_RETRY_AFTER_SECONDS)
     except ValueError:
-        # RFC 7231 allows HTTP-date format (e.g. "Wed, 21 Oct 2015 07:28:00 GMT")
-        # but we don't parse it; fall back to counted backoff.
+        pass
+
+    # Try HTTP-date format (RFC 7231 §7.1.1.1)
+    try:
+        target_dt = parsedate_to_datetime(retry_after)
+        delay = int(target_dt.timestamp() - _time.time())
+        if delay <= 0:
+            return None
+        return min(delay, MAX_RETRY_AFTER_SECONDS)
+    except (TypeError, ValueError, OverflowError):
         log = logging.getLogger('segment')
         log.warning('Unrecognized Retry-After format %r; ignoring header.', retry_after)
         return None

diff --git a/segment/analytics/test/test_client.py b/segment/analytics/test/test_client.py
@@ -359,4 +359,28 @@ def mock_post_fn(*args, **kwargs):
             mock_post.assert_called_once()
             args, kwargs = mock_post.call_args
             self.assertIn('proxies', kwargs)
-            self.assertEqual(kwargs['proxies'], proxies)
+            self.assertEqual(kwargs['proxies'], proxies)
+
+    def test_queue_full_returns_false(self):
+        """track() returns (False, msg) when the queue is full — caller should dead-letter"""
+        client = Client('testsecret', max_queue_size=1)
+
+        # Fill the queue
+        client.track('user-1', 'First Event')
+
+        # This one should be rejected
+        success, msg = client.track('user-2', 'Overflow Event')
+
+        self.assertFalse(success)
+        self.assertEqual(msg['event'], 'Overflow Event')
+
+    def test_queue_full_does_not_raise(self):
+        """track() never raises when the queue is full — returns False silently"""
+        client = Client('testsecret', max_queue_size=1)
+        client.track('user-1', 'First Event')
+
+        try:
+            success, _ = client.track('user-2', 'Overflow Event')
+            self.assertFalse(success)
+        except Exception as e:
+            self.fail(f'track() raised unexpectedly on full queue: {e}')
diff --git a/segment/analytics/test/test_consumer.py b/segment/analytics/test/test_consumer.py
@@ -384,8 +384,8 @@ def mock_post_fn(*args, **kwargs):
         self.assertLessEqual(consumer.rate_limited_until, now + 310)
         self.assertGreater(consumer.rate_limited_until, now + 290)
 
-    def test_408_and_503_use_backoff(self):
-        """Test that 408 and 503 use exponential backoff"""
+    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'}
 
         for status_code in [408, 503]:
@@ -398,7 +398,7 @@ def mock_post_fn(*args, **kwargs):
                 call_count += 1
                 if call_count == 1:
                     response = mock.Mock()
-                    response.headers = {'Retry-After': '5'}
+                    response.headers = {}  # No Retry-After
                     error = APIError(status_code, 'error', 'Error')
                     error.response = response
                     raise error
@@ -411,10 +411,54 @@ def mock_sleep(duration):
                 with mock.patch('time.sleep', side_effect=mock_sleep):
                     consumer.request([track])
 
-            # Should use backoff delay (0 for first retry), NOT the Retry-After value of 5
+            # Should use backoff delay (0 for first retry), not Retry-After
             self.assertEqual(call_count, 2)
             self.assertEqual(len(sleep_durations), 1)
-            self.assertEqual(sleep_durations[0], 0, f'{status_code} should use backoff, not Retry-After')
+            self.assertEqual(sleep_durations[0], 0, f'{status_code} without Retry-After should use backoff')
+
+    def test_503_with_retry_after_sets_rate_limit_state(self):
+        """503 with Retry-After > 0 blocks the pipeline (sets rate_limit_state) and raises"""
+        consumer = Consumer(None, 'testsecret', retries=2)
+        track = {'type': 'track', 'event': 'python event', 'userId': 'userId'}
+
+        def mock_post_fn(*args, **kwargs):
+            response = mock.Mock()
+            response.headers = {'Retry-After': '2'}
+            error = APIError(503, 'unavailable', 'Service Unavailable')
+            error.response = response
+            raise error
+
+        with mock.patch('segment.analytics.consumer.post', side_effect=mock_post_fn):
+            with self.assertRaises(APIError) as ctx:
+                consumer.request([track])
+            self.assertEqual(ctx.exception.status, 503)
+
+        # Rate-limit state should be set (pipeline-blocking)
+        self.assertIsNotNone(consumer.rate_limited_until)
+        self.assertIsNotNone(consumer.rate_limit_start_time)
+        self.assertGreater(consumer.rate_limited_until, time.time())
+
+    def test_529_with_retry_after_sets_rate_limit_state(self):
+        """529 with Retry-After > 0 blocks the pipeline (sets rate_limit_state) and raises"""
+        consumer = Consumer(None, 'testsecret', retries=2)
+        track = {'type': 'track', 'event': 'python event', 'userId': 'userId'}
+
+        def mock_post_fn(*args, **kwargs):
+            response = mock.Mock()
+            response.headers = {'Retry-After': '3'}
+            error = APIError(529, 'too_many_requests', 'Too Many Requests')
+            error.response = response
+            raise error
+
+        with mock.patch('segment.analytics.consumer.post', side_effect=mock_post_fn):
+            with self.assertRaises(APIError) as ctx:
+                consumer.request([track])
+            self.assertEqual(ctx.exception.status, 529)
+
+        # Rate-limit state should be set (pipeline-blocking)
+        self.assertIsNotNone(consumer.rate_limited_until)
+        self.assertIsNotNone(consumer.rate_limit_start_time)
+        self.assertGreater(consumer.rate_limited_until, time.time())
 
     def test_exponential_backoff_with_jitter(self):
         """Test that exponential backoff is used for retries without Retry-After"""
@@ -982,19 +1026,15 @@ def mock_post_fn(*args, **kwargs):
         # so it raises on the very first failure (1 attempt total).
         self.assertEqual(call_count, 1)
 
-    def test_parse_retry_after_http_date_logs_warning(self):
-        """parse_retry_after logs a warning for HTTP-date format and returns None"""
-        import logging
+    def test_parse_retry_after_http_date_in_past_returns_none(self):
+        """parse_retry_after returns None for an HTTP-date in the past"""
         from segment.analytics.request import parse_retry_after
 
         response = mock.Mock()
         response.headers = {'Retry-After': 'Wed, 21 Oct 2015 07:28:00 GMT'}
 
-        with self.assertLogs('segment', level=logging.WARNING) as cm:
-            result = parse_retry_after(response)
-
+        result = parse_retry_after(response)
         self.assertIsNone(result)
-        self.assertTrue(any('Unrecognized Retry-After' in line for line in cm.output))
 
     def test_410_and_460_retried(self):
         """410 and 460 are retryable status codes"""
@@ -1033,3 +1073,29 @@ def mock_post_fn(*args, **kwargs):
             self.assertEqual(ctx.exception.status, 505)
 
         self.assertEqual(call_count, 1)
+
+    def test_retries_exhausted_calls_on_error(self):
+        """on_error is called with the batch when all retries are exhausted"""
+        q = Queue()
+        on_error_calls = []
+
+        def on_error(error, batch):
+            on_error_calls.append((error, batch))
+
+        consumer = Consumer(q, 'testsecret', retries=2, on_error=on_error)
+        track = {'type': 'track', 'event': 'test event', 'userId': 'user-1'}
+        q.put(track)
+
+        def mock_post_fn(*args, **kwargs):
+            raise APIError(500, 'error', 'Server Error')
+
+        with mock.patch('segment.analytics.consumer.post', side_effect=mock_post_fn):
+            with mock.patch('time.sleep'):
+                consumer.upload()
+
+        self.assertEqual(len(on_error_calls), 1)
+        error, batch = on_error_calls[0]
+        self.assertIsInstance(error, APIError)
+        self.assertEqual(error.status, 500)
+        self.assertEqual(len(batch), 1)
+        self.assertEqual(batch[0]['event'], 'test event')
diff --git a/segment/analytics/test/test_request.py b/segment/analytics/test/test_request.py
@@ -1,4 +1,5 @@
-from datetime import datetime, date
+from datetime import datetime, date, timezone, timedelta
+import time
 import unittest
 import json
 import requests
@@ -188,12 +189,33 @@ def test_parse_retry_after_missing(self):
         self.assertIsNone(result)
 
     def test_parse_retry_after_invalid(self):
-        """Test parsing with invalid Retry-After header"""
+        """Test parsing with invalid Retry-After header (garbage string)"""
         response = mock.Mock()
         response.headers = {'Retry-After': 'invalid'}
         result = parse_retry_after(response)
         self.assertIsNone(result)
 
+    def test_parse_retry_after_http_date_future(self):
+        """Test parsing Retry-After as HTTP-date 2 seconds in future"""
+        from email.utils import format_datetime
+        future = datetime.now(tz=timezone.utc) + timedelta(seconds=2)
+        response = mock.Mock()
+        response.headers = {'Retry-After': format_datetime(future, usegmt=True)}
+        result = parse_retry_after(response)
+        # Should be approximately 2 seconds (allow 1-3 for timing)
+        self.assertIsNotNone(result)
+        self.assertGreaterEqual(result, 1)
+        self.assertLessEqual(result, 3)
+
+    def test_parse_retry_after_http_date_past(self):
+        """Test parsing Retry-After as HTTP-date in the past returns None"""
+        from email.utils import format_datetime
+        past = datetime.now(tz=timezone.utc) - timedelta(seconds=10)
+        response = mock.Mock()
+        response.headers = {'Retry-After': format_datetime(past, usegmt=True)}
+        result = parse_retry_after(response)
+        self.assertIsNone(result)
+
     def test_oauth_token_cleared_on_511(self):
         """Test that OAuth token is cleared on 511 status"""
         oauth_manager = mock.Mock()