diff --git a/README.md b/README.md
index fdcd72cf..f159b7ba 100644
--- a/README.md
+++ b/README.md
@@ -41,7 +41,7 @@ Analytics helps you measure your users, product, and business. It unlocks insigh
 
     For example, you can capture data on any app:
     ```python
-    analytics.track('Order Completed', { price: 99.84 })
+    analytics.track("Order Completed", {price: 99.84})
     ```
     Then, query the resulting data in SQL:
     ```sql
@@ -71,7 +71,7 @@ Now inside your app, you'll want to **set your** `write_key` before making any a
 ```python
 import segment.analytics as analytics
 
-analytics.write_key = 'YOUR_WRITE_KEY'
+analytics.write_key = "YOUR_WRITE_KEY"
 ```
 **Note** If you need to send data to multiple Segment sources, you can initialize a new Client for each `write_key`
 
diff --git a/e2e-cli/README.md b/e2e-cli/README.md
index 28c602a2..1a47b61f 100644
--- a/e2e-cli/README.md
+++ b/e2e-cli/README.md
@@ -2,17 +2,35 @@
 
 E2E test CLI for the [analytics-python](https://github.com/segmentio/analytics-python) SDK. Accepts a JSON input describing events and SDK configuration, sends them through the real SDK, and outputs results as JSON.
 
-## Setup
+## Running E2E tests
+
+### With devbox (recommended)
+
+```bash
+# From repo root — activates Python 3.12 and installs deps automatically
+devbox shell
+
+# Then from e2e-cli dir:
+./run-e2e.sh
+```
+
+### Without devbox
+
+Requires Python 3.9+ and Node.js 18+. Using a virtualenv is strongly recommended since macOS system Python is externally managed.
 
 ```bash
-cd e2e-cli
 python3 -m venv .venv
 source .venv/bin/activate
-pip install -r requirements.txt
-pip install -e .
+./run-e2e.sh
+```
+
+### Override sdk-e2e-tests location
+
+```bash
+E2E_TESTS_DIR=../my-e2e-tests ./run-e2e.sh
 ```
 
-## Usage
+## Manual CLI usage
 
 ```bash
 e2e-cli --input '{"writeKey":"...", ...}'
@@ -21,7 +39,7 @@ e2e-cli --input '{"writeKey":"...", ...}'
 Or without installing:
 
 ```bash
-python3 -m src.cli --input '{"writeKey":"...", ...}'
+python3 src/cli.py --input '{"writeKey":"...", ...}'
 ```
 
 ## Input Format
diff --git a/e2e-cli/e2e-config.json b/e2e-cli/e2e-config.json
index b0ccf30c..e1a02d5b 100644
--- a/e2e-cli/e2e-config.json
+++ b/e2e-cli/e2e-config.json
@@ -1,6 +1,6 @@
 {
   "sdk": "python",
-  "test_suites": "basic",
+  "test_suites": "basic,retry",
   "auto_settings": false,
   "patch": null,
   "env": {}
diff --git a/e2e-cli/run-e2e.sh b/e2e-cli/run-e2e.sh
index 533dba9e..a90b5204 100755
--- a/e2e-cli/run-e2e.sh
+++ b/e2e-cli/run-e2e.sh
@@ -2,7 +2,9 @@
 #
 # Run E2E tests for analytics-python
 #
-# Prerequisites: Python 3, pip, Node.js 18+
+# Prerequisites: Node.js 18+ and one of:
+#   - devbox (recommended): run `devbox shell` first, then ./run-e2e.sh
+#   - Python 3.9+ with a virtualenv already activated
 #
 # Usage:
 #   ./run-e2e.sh [extra args passed to run-tests.sh]
@@ -17,15 +19,25 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 SDK_ROOT="$SCRIPT_DIR/.."
 E2E_DIR="${E2E_TESTS_DIR:-$SDK_ROOT/../sdk-e2e-tests}"
 
+# Resolve python and pip — prefer activated venv/devbox python, fall back to python3
+PYTHON="${PYTHON:-$(command -v python || command -v python3)}"
+PIP="$PYTHON -m pip"
+
+if [[ -z "$PYTHON" ]]; then
+    echo "Error: Python not found. Run 'devbox shell' first or activate a virtualenv."
+    exit 1
+fi
+
 echo "=== Building analytics-python e2e-cli ==="
+echo "Using Python: $PYTHON"
 
 # Install SDK
 cd "$SDK_ROOT"
-pip install -e .
+$PIP install -e . -q
 
 # Install e2e-cli
 cd "$SCRIPT_DIR"
-pip install -e .
+$PIP install -e . -q
 
 echo ""
 
diff --git a/e2e-cli/src/cli.py b/e2e-cli/src/cli.py
index dc74d150..08d07eab 100644
--- a/e2e-cli/src/cli.py
+++ b/e2e-cli/src/cli.py
@@ -71,6 +71,10 @@ def run(input_json: str, debug: bool):
     """Run the E2E CLI with the given input configuration."""
     logger = setup_logging(debug)
     output = {"success": False, "sentBatches": 0, "error": None}
+    delivery_errors = []
+
+    def on_error(error, batch):
+        delivery_errors.append(str(error))
 
     try:
         data = json.loads(input_json)
@@ -97,6 +101,7 @@ def run(input_json: str, debug: bool):
             write_key=write_key,
             host=api_host,
             debug=debug,
+            on_error=on_error,
             upload_size=flush_at,
             upload_interval=flush_interval,
             max_retries=max_retries,
@@ -121,10 +126,12 @@ def run(input_json: str, debug: bool):
         client.flush()
         client.join()
 
-        output["success"] = True
-        # Note: We don't have easy access to batch count from the SDK internals
-        # This could be enhanced if needed
-        output["sentBatches"] = 1  # Placeholder
+        if delivery_errors:
+            output["success"] = False
+            output["error"] = delivery_errors[0]
+        else:
+            output["success"] = True
+            output["sentBatches"] = 1
 
     except json.JSONDecodeError as e:
         output["error"] = f"Invalid JSON input: {e}"
diff --git a/pyproject.toml b/pyproject.toml
index 0db95445..e4e5ba1a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -26,7 +26,6 @@ classifiers = [
 ]
 dependencies = [
     "requests~=2.7",
-    "backoff~=2.1",
     "python-dateutil~=2.2",
     "PyJWT[crypto]~=2.12",
 ]
diff --git a/requirements.txt b/requirements.txt
index 596c10da..912848b9 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,3 @@
-backoff==2.2.1
 cryptography==44.0.0
 flake8==7.1.1
 mock==2.0.0
diff --git a/segment/analytics/client.py b/segment/analytics/client.py
index 62408d87..d5dc7e31 100644
--- a/segment/analytics/client.py
+++ b/segment/analytics/client.py
@@ -30,6 +30,8 @@ class DefaultConfig(object):
         gzip = False
         timeout = 15
         max_retries = 10
+        max_total_backoff_duration = 43200
+        max_rate_limit_duration = 43200
         proxies = None
         thread = 1
         upload_interval = 0.5
@@ -65,9 +67,16 @@ def __init__(
         oauth_key_id=DefaultConfig.oauth_key_id,
         oauth_auth_server=DefaultConfig.oauth_auth_server,
         oauth_scope=DefaultConfig.oauth_scope,
+        max_total_backoff_duration=DefaultConfig.max_total_backoff_duration,
+        max_rate_limit_duration=DefaultConfig.max_rate_limit_duration,
     ):
         require("write_key", write_key, str)
 
+        if max_total_backoff_duration is None or max_total_backoff_duration < 0:
+            raise ValueError("max_total_backoff_duration must be a non-negative number")
+        if max_rate_limit_duration is None or max_rate_limit_duration < 0:
+            raise ValueError("max_rate_limit_duration must be a non-negative number")
+
         self.queue = queue.Queue(max_queue_size)
         self.write_key = write_key
         self.on_error = on_error
@@ -78,6 +87,8 @@ def __init__(
         self.gzip = gzip
         self.timeout = timeout
         self.proxies = proxies
+        self.max_total_backoff_duration = max_total_backoff_duration
+        self.max_rate_limit_duration = max_rate_limit_duration
         self.oauth_manager = None
         if oauth_client_id and oauth_client_key and oauth_key_id:
             self.oauth_manager = OauthManager(
@@ -118,6 +129,8 @@ def __init__(
                     timeout=timeout,
                     proxies=proxies,
                     oauth_manager=self.oauth_manager,
+                    max_total_backoff_duration=max_total_backoff_duration,
+                    max_rate_limit_duration=max_rate_limit_duration,
                 )
                 self.consumers.append(consumer)
 
@@ -345,7 +358,11 @@ def _enqueue(self, msg):
             return False, msg
 
     def flush(self):
-        """Forces a flush from the internal queue to the server"""
+        """Forces a flush from the internal queue to the server.
+
+        Warning: if the consumer is currently rate-limited, this call will
+        block until the rate limit clears or max_rate_limit_duration elapses.
+        """
         queue = self.queue
         size = queue.qsize()
         queue.join()
diff --git a/segment/analytics/consumer.py b/segment/analytics/consumer.py
index 35460dda..cf09c0ef 100644
--- a/segment/analytics/consumer.py
+++ b/segment/analytics/consumer.py
@@ -1,12 +1,11 @@
 import json
 import logging
+import random
 import time
 from queue import Empty
 from threading import Thread
 
-import backoff
-
-from segment.analytics.request import APIError, DatetimeSerializer, post
+from segment.analytics.request import APIError, DatetimeSerializer, parse_retry_after, post
 
 MAX_MSG_SIZE = 32 << 10
 
@@ -14,6 +13,10 @@
 # lower to leave space for extra data that will be added later, eg. "sentAt".
 BATCH_SIZE_LIMIT = 475000
 
+# Default duration limits (12 hours in seconds)
+DEFAULT_MAX_TOTAL_BACKOFF_DURATION = 43200
+DEFAULT_MAX_RATE_LIMIT_DURATION = 43200
+
 
 class FatalError(Exception):
     def __init__(self, message):
@@ -42,6 +45,8 @@ def __init__(
         timeout=15,
         proxies=None,
         oauth_manager=None,
+        max_total_backoff_duration=DEFAULT_MAX_TOTAL_BACKOFF_DURATION,
+        max_rate_limit_duration=DEFAULT_MAX_RATE_LIMIT_DURATION,
     ):
         """Create a consumer thread."""
         Thread.__init__(self)
@@ -63,6 +68,12 @@ def __init__(
         self.timeout = timeout
         self.proxies = proxies
         self.oauth_manager = oauth_manager
+        self.max_total_backoff_duration = max_total_backoff_duration
+        self.max_rate_limit_duration = max_rate_limit_duration
+
+        # Rate-limit state
+        self.rate_limited_until = None
+        self.rate_limit_start_time = None
 
     def run(self):
         """Runs the consumer."""
@@ -76,6 +87,43 @@ def pause(self):
         """Pause the consumer."""
         self.running = False
 
+    def _wait(self, seconds):
+        """Sleep in slices so pause()/join()/atexit are not blocked for up to
+        MAX_RETRY_AFTER_SECONDS. Returns False if the consumer was stopped."""
+        deadline = time.time() + seconds
+        while self.running:
+            remaining = deadline - time.time()
+            if remaining <= 0:
+                return True
+            time.sleep(min(1.0, remaining))
+        return False
+
+    def _requeue(self, batch):
+        """Put a batch back on the queue, reporting anything that no longer fits."""
+        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 rate-limit re-queue. Dropping %d item(s).", len(dropped))
+            if self.on_error:
+                self.on_error(Exception("Queue full, items dropped during rate-limit re-queue"), dropped)
+
+    def set_rate_limit_state(self, response):
+        """Set rate-limit state from a 429 response with a valid Retry-After header."""
+        retry_after = parse_retry_after(response) if response is not None else None
+        if retry_after is not None:
+            self.rate_limited_until = time.time() + retry_after
+        if self.rate_limit_start_time is None:
+            self.rate_limit_start_time = time.time()
+
+    def clear_rate_limit_state(self):
+        """Clear rate-limit state after successful request or duration exceeded."""
+        self.rate_limited_until = None
+        self.rate_limit_start_time = None
+
     def upload(self):
         """Upload the next batch of items, return whether successful."""
         success = False
@@ -83,16 +131,64 @@ def upload(self):
         if len(batch) == 0:
             return False
 
+        # Check rate-limit state before attempting upload. Gate on the episode
+        # marker, not on rate_limited_until: the latter is cleared as soon as its
+        # wait has been served, so it cannot be used to decide whether we are
+        # still inside a rate-limit episode.
+        if self.rate_limit_start_time is not None:
+            now = time.time()
+
+            # Check if maxRateLimitDuration has been exceeded
+            if now - self.rate_limit_start_time > self.max_rate_limit_duration:
+                self.log.error(
+                    "Rate limit duration exceeded (%ds). Clearing rate-limit state and dropping batch.", self.max_rate_limit_duration
+                )
+                self.clear_rate_limit_state()
+                # Drop the batch by marking items as done
+                if self.on_error:
+                    self.on_error(Exception("Rate limit duration exceeded, batch dropped"), batch)
+                for _ in batch:
+                    self.queue.task_done()
+                return False
+
+            # Still rate-limited; wait until the rate limit expires
+            if self.rate_limited_until is not None:
+                wait_time = self.rate_limited_until - now
+                if wait_time > 0:
+                    self.log.debug("Rate-limited. Waiting %.2fs before next upload attempt.", wait_time)
+                    if not self._wait(wait_time):
+                        # Shutting down: leave the batch queued rather than
+                        # uploading into a consumer that is stopping.
+                        self._requeue(batch)
+                        return False
+                # The wait has been served. Clearing it here keeps a stale
+                # timestamp from classifying later, unrelated errors as rate limits.
+                self.rate_limited_until = None
+
         try:
             self.request(batch)
+            # Success — clear rate-limit state
+            self.clear_rate_limit_state()
             success = True
+        except APIError as e:
+            if getattr(e, "rate_limited", False):
+                self.log.debug("Rate-limited (status %d). Re-queuing batch and halting upload iteration.", e.status)
+                self._requeue(batch)
+                success = False
+            else:
+                self.log.error("error uploading: %s", e)
+                success = False
+                if self.on_error:
+                    self.on_error(e, batch)
         except Exception as e:
             self.log.error("error uploading: %s", e)
             success = False
             if self.on_error:
                 self.on_error(e, batch)
         finally:
-            # mark items as acknowledged from queue
+            # Each item in batch was obtained via queue.get() and must have
+            # exactly one matching task_done() call — including re-queued items,
+            # which will produce a new task_done() obligation on their next get().
             for _ in batch:
                 self.queue.task_done()
         return success
@@ -130,34 +226,60 @@ def next(self):
     def request(self, batch):
         """Attempt to upload the batch and retry before raising an error"""
 
-        def fatal_exception(exc):
-            if isinstance(exc, APIError):
-                # retry on server errors and client errors
-                # with 429 status code (rate limited),
-                # don't retry on other client errors
-                return (400 <= exc.status < 500) and exc.status != 429
-            elif isinstance(exc, FatalError):
+        def is_retryable_status(status):
+            # Retryable 4xx: 408, 429, 460
+            # 410 Gone: permanently removed, but included for parity with the
+            # Node.js SDK. Retrying is harmless since the server will keep
+            # returning 410, and the retry budget caps total attempts.
+            # Non-retryable 4xx: 400, 401, 403, 404, 413, 422, and all other 4xx
+            # Retryable 5xx: all except 501, 505
+            # 511: only retryable when OauthManager is configured
+            if 400 <= status < 500:
+                return status in (408, 410, 429, 460)
+            elif 500 <= status < 600:
+                if status in (501, 505):
+                    return False
+                if status == 511:
+                    return self.oauth_manager is not None
                 return True
-            else:
-                # retry on all other errors (eg. network)
-                return False
+            return False
+
+        def calculate_backoff_delay(attempt):
+            # First retry is immediate; thereafter 0.5s, 1s, 2s, 4s… capped at 60s
+            if attempt == 1:
+                return 0
+            base_delay = 0.5 * (2 ** (attempt - 2))
+            jitter = random.uniform(0, 0.1 * base_delay)
+            return min(base_delay + jitter, 60)
+
+        def apply_backoff(e, label):
+            """Apply retry backoff logic. Returns delay if should retry, raises if exhausted."""
+            nonlocal first_failure_time, backoff_attempts
+            if first_failure_time is None:
+                first_failure_time = time.time()
+            if time.time() - first_failure_time >= self.max_total_backoff_duration:
+                self.log.error(
+                    f"Max total backoff duration ({self.max_total_backoff_duration}s) exceeded "
+                    f"after {total_attempts} attempts. Final error: {e}"
+                )
+                raise e
+            backoff_attempts += 1
+            if backoff_attempts >= self.retries + 1:
+                self.log.error(f"All {self.retries} retries exhausted after {total_attempts} total attempts. Final error: {e}")
+                raise e
+            delay = calculate_backoff_delay(backoff_attempts)
+            self.log.debug(f"{label} {backoff_attempts}/{self.retries} (total attempts: {total_attempts}) after {delay:.2f}s: {e}")
+            return delay
+
+        total_attempts = 0
+        backoff_attempts = 0
+        first_failure_time = None
+
+        while True:
+            total_attempts += 1
 
-        attempt_count = 0
-
-        @backoff.on_exception(
-            backoff.expo,
-            Exception,
-            max_tries=self.retries + 1,
-            giveup=fatal_exception,
-            on_backoff=lambda details: self.log.debug(
-                f"Retry attempt {details['tries']}/{self.retries + 1} after {details['elapsed']:.2f}s"
-            ),
-        )
-        def send_request():
-            nonlocal attempt_count
-            attempt_count += 1
             try:
-                return post(
+                response = post(
                     self.write_key,
                     self.host,
                     gzip=self.gzip,
@@ -165,10 +287,34 @@ def send_request():
                     batch=batch,
                     proxies=self.proxies,
                     oauth_manager=self.oauth_manager,
+                    retry_count=total_attempts - 1,
                 )
-            except Exception as e:
-                if attempt_count >= self.retries + 1:
-                    self.log.error(f"All {self.retries} retries exhausted. Final error: {e}")
+                return response
+
+            except FatalError as e:
+                # Raised by oauth_manager when token refresh fails permanently;
+                # not safe to retry.
+                self.log.error(f"Fatal error after {total_attempts} attempts: {e}")
                 raise
 
-        send_request()
+            except APIError as e:
+                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)
+                    # Tell upload() this specific failure is a rate limit. Inferring
+                    # it from consumer state misclassifies every later error.
+                    e.rate_limited = True
+                    raise
+
+                # No Retry-After: counted backoff
+                delay = apply_backoff(e, f"Retry attempt (status {e.status})")
+                time.sleep(delay)
+
+            except Exception as e:
+                delay = apply_backoff(e, "Network error retry")
+                time.sleep(delay)
diff --git a/segment/analytics/request.py b/segment/analytics/request.py
index a163f125..8ab61670 100644
--- a/segment/analytics/request.py
+++ b/segment/analytics/request.py
@@ -1,6 +1,9 @@
+import base64
 import json
 import logging
-from datetime import date, datetime
+import time as _time
+from datetime import date, datetime, timezone
+from email.utils import parsedate_to_datetime
 from gzip import GzipFile
 from io import BytesIO
 
@@ -12,8 +15,46 @@
 
 _session = sessions.Session()
 
+# Maximum Retry-After delay to respect (5 minutes)
+MAX_RETRY_AFTER_SECONDS = 300
 
-def post(write_key, host=None, gzip=False, timeout=15, proxies=None, oauth_manager=None, **kwargs):
+
+def parse_retry_after(response):
+    """
+    Parse Retry-After header from response.
+    Returns the delay in seconds, or None if header is not present or invalid.
+    Caps the value at MAX_RETRY_AFTER_SECONDS.
+    """
+    retry_after = response.headers.get("Retry-After")
+    if not retry_after:
+        return None
+
+    try:
+        delay = int(retry_after)
+        return min(max(delay, 0), MAX_RETRY_AFTER_SECONDS)
+    except ValueError:
+        pass
+
+    # Try HTTP-date format (RFC 7231 §7.1.1.1)
+    try:
+        target_dt = parsedate_to_datetime(retry_after)
+        if target_dt.tzinfo is None:
+            # parsedate_to_datetime returns a naive datetime for the RFC 5322
+            # "-0000" offset, which servers do emit. timestamp() would then read
+            # it in the host's local zone, so the same header yields different
+            # delays — or None — depending on where the process runs.
+            target_dt = target_dt.replace(tzinfo=timezone.utc)
+        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
+
+
+def post(write_key, host=None, gzip=False, timeout=15, proxies=None, oauth_manager=None, retry_count=0, **kwargs):
     """Post the `kwargs` to the API"""
     log = logging.getLogger("segment")
     body = kwargs
@@ -26,9 +67,21 @@ def post(write_key, host=None, gzip=False, timeout=15, proxies=None, oauth_manag
         auth = oauth_manager.get_token()
     data = json.dumps(body, cls=DatetimeSerializer)
     log.debug("making request: %s", data)
-    headers = {"Content-Type": "application/json", "User-Agent": "analytics-python/" + VERSION}
+    headers = {
+        "Content-Type": "application/json",
+        "User-Agent": "analytics-python/" + VERSION,
+    }
+    if retry_count > 0:
+        headers["X-Retry-Count"] = str(retry_count)
+
+    # Add Authorization header - prefer OAuth Bearer token, fallback to Basic auth
     if auth:
         headers["Authorization"] = "Bearer {}".format(auth)
+    else:
+        # Basic auth with write key (format: "writeKey:" encoded in base64)
+        credentials = "{}:".format(write_key)
+        encoded = base64.b64encode(credentials.encode("utf-8")).decode("utf-8")
+        headers["Authorization"] = "Basic {}".format(encoded)
 
     if gzip:
         headers["Content-Encoding"] = "gzip"
@@ -53,27 +106,28 @@ def post(write_key, host=None, gzip=False, timeout=15, proxies=None, oauth_manag
     except Exception as e:
         raise e
 
-    if res.status_code == 200:
+    if 200 <= res.status_code < 400:
         log.debug("data uploaded successfully")
         return res
 
-    if oauth_manager and res.status_code in [400, 401, 403]:
+    if oauth_manager and res.status_code in [400, 401, 403, 511]:
         oauth_manager.clear_token()
 
     try:
         payload = res.json()
         log.debug("received response: %s", payload)
-        raise APIError(res.status_code, payload["code"], payload["message"])
-    except ValueError:
+        raise APIError(res.status_code, payload["code"], payload["message"], res)
+    except (ValueError, KeyError):
         log.error("Unknown error: [%s] %s", res.status_code, res.reason)
-        raise APIError(res.status_code, "unknown", res.text)
+        raise APIError(res.status_code, "unknown", res.text, res)
 
 
 class APIError(Exception):
-    def __init__(self, status, code, message):
+    def __init__(self, status, code, message, response=None):
         self.message = message
         self.status = status
         self.code = code
+        self.response = response
 
     def __str__(self):
         msg = "[Segment] {0}: {1} ({2})"
diff --git a/segment/analytics/test/test_client.py b/segment/analytics/test/test_client.py
index e71d22d3..68d6bd78 100644
--- a/segment/analytics/test/test_client.py
+++ b/segment/analytics/test/test_client.py
@@ -367,3 +367,32 @@ def mock_post_fn(*args, **kwargs):
             args, kwargs = mock_post.call_args
             self.assertIn("proxies", kwargs)
             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)
+        # Ensure consumer thread is no longer uploading
+        client.join()
+
+        # 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)
+        # Ensure consumer thread is no longer uploading
+        client.join()
+
+        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
index 2d45c8b8..c7115dcc 100644
--- a/segment/analytics/test/test_consumer.py
+++ b/segment/analytics/test/test_consumer.py
@@ -1,4 +1,5 @@
 import json
+import threading
 import time
 import unittest
 
@@ -9,7 +10,7 @@
 except ImportError:
     from Queue import Queue
 
-from segment.analytics.consumer import MAX_MSG_SIZE, Consumer
+from segment.analytics.consumer import MAX_MSG_SIZE, Consumer, FatalError
 from segment.analytics.request import APIError
 
 
@@ -118,7 +119,7 @@ def test_request_retry(self):
         consumer = Consumer(None, "testsecret")
         self._test_request_retry(consumer, APIError(500, "code", "Internal Server Error"), 2)
 
-        # we should retry on HTTP 429 errors
+        # 429 without Retry-After uses counted backoff (like other retryable errors)
         consumer = Consumer(None, "testsecret")
         self._test_request_retry(consumer, APIError(429, "code", "Too Many Requests"), 2)
 
@@ -128,7 +129,7 @@ def test_request_retry(self):
         try:
             self._test_request_retry(consumer, api_error, 1)
         except APIError:
-            pass
+            pass  # Expected: 400 is non-retryable, so the error propagates here
         else:
             self.fail("request() should not retry on client errors")
 
@@ -180,3 +181,920 @@ def mock_post_fn(*args, **kwargs):
             args, kwargs = mock_post.call_args
             cls().assertIn("proxies", kwargs)
             cls().assertEqual(kwargs["proxies"], proxies)
+
+    def test_retry_count_header_increments(self):
+        """Test that X-Retry-Count header increments on each retry"""
+        consumer = Consumer(None, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        retry_counts = []
+
+        def mock_post_fn(*args, **kwargs):
+            retry_counts.append(kwargs.get("retry_count", 0))
+            if len(retry_counts) < 3:
+                raise APIError(500, "error", "Server Error")
+            # Success on third attempt
+            return mock.Mock(status_code=200)
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            consumer.request([track])
+
+        # Should have been called 3 times with retry counts 0, 1, 2
+        self.assertEqual(retry_counts, [0, 1, 2])
+
+    def test_non_retryable_4xx_status_codes(self):
+        """Test that non-retryable 4xx errors are not retried"""
+        consumer = Consumer(None, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        non_retryable_codes = [400, 401, 403, 404, 413, 422]
+
+        for status_code in non_retryable_codes:
+            call_count = 0
+
+            def mock_post_fn(*args, **kwargs):
+                nonlocal call_count
+                call_count += 1
+                raise APIError(status_code, "error", f"Client Error {status_code}")
+
+            with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+                try:
+                    consumer.request([track])
+                except APIError as e:
+                    self.assertEqual(e.status, status_code)
+
+                # Should only be called once (no retries)
+                self.assertEqual(call_count, 1, f"Status {status_code} should not be retried")
+
+    def test_retryable_4xx_status_codes(self):
+        """Test that retryable 4xx errors are retried (429 without Retry-After uses backoff too)"""
+        consumer = Consumer(None, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        retryable_codes = [408, 410, 429, 460]
+
+        for status_code in retryable_codes:
+            call_count = 0
+
+            def mock_post_fn(*args, **kwargs):
+                nonlocal call_count
+                call_count += 1
+                if call_count < 3:
+                    raise APIError(status_code, "error", f"Retryable Error {status_code}")
+                # Success on third attempt
+                return mock.Mock(status_code=200)
+
+            with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+                with mock.patch("time.sleep"):  # Mock sleep to speed up test
+                    consumer.request([track])
+
+                # Should have been called 3 times
+                self.assertEqual(call_count, 3, f"Status {status_code} should be retried")
+
+    def test_non_retryable_5xx_status_codes(self):
+        """Test that non-retryable 5xx errors are not retried"""
+        consumer = Consumer(None, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        non_retryable_codes = [501, 505]
+
+        for status_code in non_retryable_codes:
+            call_count = 0
+
+            def mock_post_fn(*args, **kwargs):
+                nonlocal call_count
+                call_count += 1
+                raise APIError(status_code, "error", f"Server Error {status_code}")
+
+            with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+                try:
+                    consumer.request([track])
+                except APIError as e:
+                    self.assertEqual(e.status, status_code)
+
+                # Should only be called once (no retries)
+                self.assertEqual(call_count, 1, f"Status {status_code} should not be retried")
+
+    def test_retryable_5xx_status_codes(self):
+        """Test that retryable 5xx errors are retried"""
+        consumer = Consumer(None, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        retryable_codes = [500, 502, 503, 504]
+
+        for status_code in retryable_codes:
+            call_count = 0
+
+            def mock_post_fn(*args, **kwargs):
+                nonlocal call_count
+                call_count += 1
+                if call_count < 3:
+                    raise APIError(status_code, "error", f"Server Error {status_code}")
+                # Success on third attempt
+                return mock.Mock(status_code=200)
+
+            with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+                with mock.patch("time.sleep"):  # Mock sleep to speed up test
+                    consumer.request([track])
+
+                # Should have been called 3 times
+                self.assertEqual(call_count, 3, f"Status {status_code} should be retried")
+
+    def test_429_sets_rate_limit_state_with_retry_after(self):
+        """Test that 429 with Retry-After sets rate_limited_until on consumer"""
+        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": "10"}
+            error = APIError(429, "rate_limit", "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, 429)
+
+        # Rate-limit state should be set
+        self.assertIsNotNone(consumer.rate_limited_until)
+        self.assertIsNotNone(consumer.rate_limit_start_time)
+        # rate_limited_until should be ~10 seconds in the future
+        self.assertGreater(consumer.rate_limited_until, time.time() + 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"""
+        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": "600"}  # 10 minutes
+            error = APIError(429, "rate_limit", "Too Many Requests")
+            error.response = response
+            raise error
+
+        now = time.time()
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with self.assertRaises(APIError):
+                consumer.request([track])
+
+        # 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_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]:
+            consumer = Consumer(None, "testsecret", retries=2)
+            call_count = 0
+            sleep_durations = []
+
+            def mock_post_fn(*args, **kwargs):
+                nonlocal call_count
+                call_count += 1
+                if call_count == 1:
+                    response = mock.Mock()
+                    response.headers = {}  # No Retry-After
+                    error = APIError(status_code, "error", "Error")
+                    error.response = response
+                    raise error
+                return mock.Mock(status_code=200)
+
+            def mock_sleep(duration):
+                sleep_durations.append(duration)
+
+            with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+                with mock.patch("time.sleep", side_effect=mock_sleep):
+                    consumer.request([track])
+
+            # 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} 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_stale_rate_limit_state_does_not_misroute_later_errors(self):
+        """A past rate-limit episode must not make later non-retryable errors look rate-limited"""
+        q = Queue()
+        consumer = Consumer(q, "testsecret", retries=1)
+        consumer.on_error = mock.Mock()
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+        q.put(track)
+
+        # Simulate having been rate-limited a moment ago and already served the wait.
+        consumer.rate_limit_start_time = time.time() - 1
+        consumer.rate_limited_until = time.time() - 0.5
+
+        def mock_post_fn(*args, **kwargs):
+            raise APIError(400, "bad_request", "Bad Request")
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            consumer.upload()
+
+        # The 400 is non-retryable: it must be dropped and reported, not re-queued.
+        self.assertTrue(consumer.on_error.called)
+        self.assertEqual(q.qsize(), 0)
+
+    def test_rate_limit_wait_is_interruptible(self):
+        """pause() must break the Retry-After wait rather than blocking for its full duration"""
+        consumer = Consumer(Queue(), "testsecret")
+
+        def stop_soon():
+            time.sleep(0.2)
+            consumer.pause()
+
+        t = threading.Thread(target=stop_soon)
+        t.start()
+        started = time.time()
+        completed = consumer._wait(30)
+        elapsed = time.time() - started
+        t.join()
+
+        self.assertFalse(completed, "the wait should report that it was interrupted")
+        self.assertLess(elapsed, 5, f"pause() did not interrupt the wait; it took {elapsed:.1f}s")
+
+    def test_exponential_backoff_with_jitter(self):
+        """Test that exponential backoff is used for retries without Retry-After"""
+        consumer = Consumer(None, "testsecret", retries=4)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        call_count = 0
+        sleep_durations = []
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+
+            if call_count <= 3:
+                raise APIError(500, "error", "Server Error")
+
+            return mock.Mock(status_code=200)
+
+        def mock_sleep(duration):
+            sleep_durations.append(duration)
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with mock.patch("time.sleep", side_effect=mock_sleep):
+                consumer.request([track])
+
+        # Should have 3 backoff delays
+        self.assertEqual(len(sleep_durations), 3)
+
+        # Delays should be increasing (exponential)
+        # First: 0s (immediate), Second: ~0.5s, Third: ~1s (with jitter)
+        self.assertEqual(sleep_durations[0], 0)  # First retry is immediate
+        self.assertGreater(sleep_durations[1], 0.4)
+        self.assertLess(sleep_durations[1], 0.6)
+        self.assertGreater(sleep_durations[2], 0.9)
+        self.assertLess(sleep_durations[2], 1.2)
+
+    def test_fatal_error_not_retried(self):
+        """Test that FatalError is not retried"""
+        consumer = Consumer(None, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        call_count = 0
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            raise FatalError("Fatal error occurred")
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with self.assertRaises(FatalError):
+                consumer.request([track])
+
+        # Should only be called once (no retries)
+        self.assertEqual(call_count, 1)
+
+    def test_max_retries_exhausted(self):
+        """Test that request fails after max retries exhausted"""
+        consumer = Consumer(None, "testsecret", retries=2)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        call_count = 0
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            # Always fail with retryable error
+            raise APIError(500, "error", "Server Error")
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with mock.patch("time.sleep"):  # Mock sleep to speed up test
+                try:
+                    consumer.request([track])
+                except APIError as e:
+                    self.assertEqual(e.status, 500)
+
+        # Should be called 3 times (initial + 2 retries)
+        self.assertEqual(call_count, 3)
+
+    def test_first_request_has_retry_count_zero(self):
+        """T01: First successful request includes X-Retry-Count=0"""
+        consumer = Consumer(None, "testsecret")
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        retry_count = None
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal retry_count
+            retry_count = kwargs.get("retry_count")
+            return mock.Mock(status_code=200)
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            consumer.request([track])
+
+        # First request should have retry_count=0
+        self.assertEqual(retry_count, 0)
+
+    def test_429_without_retry_after_uses_counted_backoff(self):
+        """429 without Retry-After uses counted backoff (not pipeline blocking)"""
+        consumer = Consumer(None, "testsecret", retries=2)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        call_count = 0
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count < 3:
+                error = APIError(429, "rate_limit", "Too Many Requests")
+                error.response = mock.Mock()
+                error.response.headers = {}  # No Retry-After
+                raise error
+            return mock.Mock(status_code=200)
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with mock.patch("time.sleep"):
+                consumer.request([track])
+
+        # Should retry with backoff (3 calls: initial + 2 retries)
+        self.assertEqual(call_count, 3)
+        # Rate-limit state should NOT be set (no pipeline blocking)
+        self.assertIsNone(consumer.rate_limited_until)
+
+    def test_408_without_retry_after_uses_backoff(self):
+        """T10: 408 without Retry-After header uses backoff retry"""
+        consumer = Consumer(None, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        call_count = 0
+        retry_counts = []
+        sleep_duration = None
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            retry_counts.append(kwargs.get("retry_count", 0))
+
+            if call_count == 1:
+                # 408 without Retry-After header
+                error = APIError(408, "timeout", "Request Timeout")
+                error.response = mock.Mock()
+                error.response.headers = {}  # No Retry-After
+                raise error
+
+            return mock.Mock(status_code=200)
+
+        def mock_sleep(duration):
+            nonlocal sleep_duration
+            sleep_duration = duration
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with mock.patch("time.sleep", side_effect=mock_sleep):
+                consumer.request([track])
+
+        # Should have two attempts
+        self.assertEqual(call_count, 2)
+        self.assertEqual(retry_counts, [0, 1])
+
+        # First retry should be immediate (0s delay)
+        self.assertIsNotNone(sleep_duration)
+        if sleep_duration is not None:
+            self.assertEqual(sleep_duration, 0)
+
+    def test_network_error_retried_with_backoff(self):
+        """T15: Network/IO error is retried with backoff"""
+        consumer = Consumer(None, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        call_count = 0
+        retry_counts = []
+        sleep_duration = None
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            retry_counts.append(kwargs.get("retry_count", 0))
+
+            if call_count == 1:
+                # Network error
+                raise ConnectionError("Network connection failed")
+
+            return mock.Mock(status_code=200)
+
+        def mock_sleep(duration):
+            nonlocal sleep_duration
+            sleep_duration = duration
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with mock.patch("time.sleep", side_effect=mock_sleep):
+                consumer.request([track])
+
+        # Should have two attempts
+        self.assertEqual(call_count, 2)
+        self.assertEqual(retry_counts, [0, 1])
+
+        # First retry should be immediate (0s delay)
+        self.assertIsNotNone(sleep_duration)
+        if sleep_duration is not None:
+            self.assertEqual(sleep_duration, 0)
+
+    def test_511_not_retryable_without_oauth(self):
+        """T17: 511 is NOT retried when OauthManager is not configured"""
+        consumer = Consumer(None, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        call_count = 0
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            raise APIError(511, "auth_required", "Network Authentication Required")
+
+        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, 511)
+
+        # Should only be called once (not retried without OAuth)
+        self.assertEqual(call_count, 1)
+
+    def test_511_retryable_with_oauth(self):
+        """T17: 511 IS retried when OauthManager is configured"""
+        oauth_manager = mock.Mock()
+        consumer = Consumer(None, "testsecret", retries=3, oauth_manager=oauth_manager)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        call_count = 0
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count < 3:
+                raise APIError(511, "auth_required", "Network Authentication Required")
+            return mock.Mock(status_code=200)
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with mock.patch("time.sleep"):
+                consumer.request([track])
+
+        # Should have been called 3 times (511 is retryable with OAuth)
+        self.assertEqual(call_count, 3)
+
+    def test_429_with_retry_after_does_not_count_against_backoff_budget(self):
+        """429 with Retry-After raises immediately (pipeline blocking) without consuming backoff budget"""
+        consumer = Consumer(None, "testsecret", retries=1)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        call_count = 0
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            error = APIError(429, "rate_limit", "Too Many Requests")
+            error.response = mock.Mock()
+            error.response.headers = {"Retry-After": "1"}
+            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, 429)
+
+        # 429 with Retry-After raises on first attempt (pipeline blocking)
+        self.assertEqual(call_count, 1)
+
+    def test_413_payload_too_large_not_retried(self):
+        """T12: 413 Payload Too Large is non-retryable (won't succeed on retry)"""
+        consumer = Consumer(None, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        call_count = 0
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            raise APIError(413, "payload_too_large", "Payload Too Large")
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            try:
+                consumer.request([track])
+            except APIError as e:
+                self.assertEqual(e.status, 413)
+
+        # Should only be called once (no retries)
+        self.assertEqual(call_count, 1)
+
+    def test_t04_429_halts_upload_iteration(self):
+        """T04: 429 halts current upload iteration — batch is re-queued, not dropped"""
+        q = Queue()
+        consumer = Consumer(q, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        # Put a message in the queue
+        q.put(track)
+
+        call_count = 0
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            response = mock.Mock()
+            response.headers = {"Retry-After": "10"}
+            error = APIError(429, "rate_limit", "Too Many Requests")
+            error.response = response
+            raise error
+
+        on_error_called = []
+
+        def on_error(e, batch):
+            on_error_called.append((e, batch))
+
+        consumer.on_error = on_error
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with mock.patch("time.sleep"):
+                result = consumer.upload()
+
+        # upload() should return False (not successful)
+        self.assertFalse(result)
+        # request() should have been called exactly once
+        self.assertEqual(call_count, 1)
+        # on_error should NOT have been called (batch was re-queued, not dropped)
+        self.assertEqual(len(on_error_called), 0)
+        # Rate-limit state should be set
+        self.assertIsNotNone(consumer.rate_limited_until)
+        self.assertIsNotNone(consumer.rate_limit_start_time)
+
+    def test_429_without_retry_after_does_not_requeue_batch(self):
+        """429 without Retry-After is treated as normal failure in upload() and is not re-queued"""
+        q = Queue()
+        consumer = Consumer(q, "testsecret", retries=0)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+        q.put(track)
+
+        def mock_post_fn(*args, **kwargs):
+            error = APIError(429, "rate_limit", "Too Many Requests")
+            error.response = mock.Mock()
+            error.response.headers = {}
+            raise error
+
+        on_error_called = []
+
+        def on_error(e, batch):
+            on_error_called.append((e, batch))
+
+        consumer.on_error = on_error
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with mock.patch("time.sleep"):
+                result = consumer.upload()
+
+        self.assertFalse(result)
+        self.assertEqual(len(on_error_called), 1)
+        self.assertIsNone(consumer.rate_limited_until)
+        self.assertEqual(q.qsize(), 0)
+
+    def test_retry_after_zero_uses_counted_backoff(self):
+        """429 with Retry-After: 0 falls through to counted backoff (not pipeline blocking).
+        Prevents a tight re-queue loop when the server says retry immediately."""
+        consumer = Consumer(None, "testsecret", retries=2)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        call_count = 0
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count < 3:
+                response = mock.Mock()
+                response.headers = {"Retry-After": "0"}
+                error = APIError(429, "rate_limit", "Too Many Requests")
+                error.response = response
+                raise error
+            return mock.Mock(status_code=200)
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with mock.patch("time.sleep"):
+                consumer.request([track])
+
+        # Should retry with backoff (3 calls: initial + 2 retries)
+        self.assertEqual(call_count, 3)
+        # Rate-limit state must NOT be set (no pipeline blocking for Retry-After: 0)
+        self.assertIsNone(consumer.rate_limited_until)
+
+    def test_t19_max_total_backoff_duration(self):
+        """T19: Gives up after maxTotalBackoffDuration elapsed"""
+        consumer = Consumer(None, "testsecret", retries=1000, max_total_backoff_duration=5)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        call_count = 0
+        fake_time = [100.0]  # Start time
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            raise APIError(500, "error", "Server Error")
+
+        def mock_time():
+            # Advance time by 3 seconds on each call after the first
+            result = fake_time[0]
+            fake_time[0] += 3.0
+            return result
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with mock.patch("time.sleep"):
+                with mock.patch("time.time", side_effect=mock_time):
+                    with self.assertRaises(APIError) as ctx:
+                        consumer.request([track])
+                    self.assertEqual(ctx.exception.status, 500)
+
+        # With max_total_backoff_duration=5 and time advancing 3s per call:
+        # Attempt 1: fails, first_failure_time set at 100, time now 103
+        # Attempt 2: fails, time is 106, 106-100=6 > 5, exceeds duration
+        # So should be called exactly 2 times
+        self.assertEqual(call_count, 2)
+
+    def test_t20_max_rate_limit_duration(self):
+        """T20: Rate-limited state clears and batch is dropped after maxRateLimitDuration"""
+        q = Queue()
+        consumer = Consumer(q, "testsecret", retries=3, max_rate_limit_duration=10)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        # Pre-set rate-limit state as if we entered it 15 seconds ago
+        now = time.time()
+        consumer.rate_limit_start_time = now - 15  # 15s ago, exceeds 10s limit
+        consumer.rate_limited_until = now + 5  # Would still be rate-limited
+
+        # Put a message in the queue
+        q.put(track)
+
+        on_error_called = []
+
+        def on_error(e, batch):
+            on_error_called.append((e, batch))
+
+        consumer.on_error = on_error
+
+        # upload() should detect duration exceeded, clear state, drop batch
+        result = consumer.upload()
+
+        self.assertFalse(result)
+        # Rate-limit state should be cleared
+        self.assertIsNone(consumer.rate_limited_until)
+        self.assertIsNone(consumer.rate_limit_start_time)
+        # on_error should have been called (batch was dropped)
+        self.assertEqual(len(on_error_called), 1)
+
+    def test_rate_limit_state_cleared_on_success(self):
+        """Rate-limit state is cleared after a successful request"""
+        q = Queue()
+        consumer = Consumer(q, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        # Set rate-limit state
+        consumer.rate_limited_until = time.time() - 1  # Already expired
+        consumer.rate_limit_start_time = time.time() - 10
+
+        q.put(track)
+
+        with mock.patch("segment.analytics.consumer.post", return_value=mock.Mock(status_code=200)):
+            result = consumer.upload()
+
+        self.assertTrue(result)
+        # Rate-limit state should be cleared on success
+        self.assertIsNone(consumer.rate_limited_until)
+        self.assertIsNone(consumer.rate_limit_start_time)
+
+    def test_retry_after_zero_does_not_trigger_pipeline_blocking(self):
+        """Retry-After: 0 must not cause tight re-queue loop; falls through to counted backoff"""
+        q = Queue()
+        consumer = Consumer(q, "testsecret", retries=2)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+        q.put(track)
+
+        call_count = 0
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                response = mock.Mock()
+                response.headers = {"Retry-After": "0"}
+                error = APIError(429, "rate_limit", "Too Many Requests")
+                error.response = response
+                raise error
+            return mock.Mock(status_code=200)
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with mock.patch("time.sleep"):
+                result = consumer.upload()
+
+        # Should succeed on retry (counted backoff path, not pipeline-blocking)
+        self.assertTrue(result)
+        # Rate-limit state must NOT be set (no pipeline blocking)
+        self.assertIsNone(consumer.rate_limited_until)
+
+    def test_queue_full_during_429_requeue_calls_on_error(self):
+        """Queue-full during 429 re-queue calls on_error with dropped items"""
+        from queue import Full
+
+        q = Queue()
+        consumer = Consumer(q, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+        q.put(track)
+
+        dropped_batches = []
+
+        def on_error(e, batch):
+            dropped_batches.append(batch)
+
+        consumer.on_error = on_error
+
+        def mock_post_fn(*args, **kwargs):
+            response = mock.Mock()
+            response.headers = {"Retry-After": "5"}
+            error = APIError(429, "rate_limit", "Too Many Requests")
+            error.response = response
+            raise error
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with mock.patch("time.sleep"):
+                # Make queue.put raise Full to simulate a full queue
+                with mock.patch.object(q, "put", side_effect=Full):
+                    result = consumer.upload()
+
+        self.assertFalse(result)
+        # on_error should have been called with the dropped item
+        self.assertEqual(len(dropped_batches), 1)
+        self.assertEqual(len(dropped_batches[0]), 1)
+
+    def test_none_max_total_backoff_duration_rejected_by_client(self):
+        """Client rejects None for max_total_backoff_duration"""
+        from segment.analytics.client import Client
+
+        with self.assertRaises(ValueError):
+            Client("testsecret", max_total_backoff_duration=None)
+
+    def test_none_max_rate_limit_duration_rejected_by_client(self):
+        """Client rejects None for max_rate_limit_duration"""
+        from segment.analytics.client import Client
+
+        with self.assertRaises(ValueError):
+            Client("testsecret", max_rate_limit_duration=None)
+
+    def test_max_total_backoff_duration_zero_prevents_retry(self):
+        """max_total_backoff_duration=0 prevents any retry attempt"""
+        consumer = Consumer(None, "testsecret", retries=1000, max_total_backoff_duration=0)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+
+        call_count = 0
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            raise APIError(500, "error", "Server Error")
+
+        with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+            with mock.patch("time.sleep"):
+                with self.assertRaises(APIError):
+                    consumer.request([track])
+
+        # With duration=0 and >= check, first failure sets first_failure_time
+        # and immediately satisfies time.time() - first_failure_time >= 0,
+        # so it raises on the very first failure (1 attempt total).
+        self.assertEqual(call_count, 1)
+
+    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"}
+
+        result = parse_retry_after(response)
+        self.assertIsNone(result)
+
+    def test_410_and_460_retried(self):
+        """410 and 460 are retryable status codes"""
+        for status_code in [410, 460]:
+            consumer = Consumer(None, "testsecret", retries=2)
+            track = {"type": "track", "event": "python event", "userId": "userId"}
+            call_count = 0
+
+            def mock_post_fn(*args, _status=status_code, **kwargs):
+                nonlocal call_count
+                call_count += 1
+                if call_count < 2:
+                    raise APIError(_status, "error", f"Error {_status}")
+                return mock.Mock(status_code=200)
+
+            with mock.patch("segment.analytics.consumer.post", side_effect=mock_post_fn):
+                with mock.patch("time.sleep"):
+                    consumer.request([track])
+
+            self.assertEqual(call_count, 2, f"{status_code} should be retried")
+
+    def test_505_not_retried(self):
+        """505 HTTP Version Not Supported is non-retryable"""
+        consumer = Consumer(None, "testsecret", retries=3)
+        track = {"type": "track", "event": "python event", "userId": "userId"}
+        call_count = 0
+
+        def mock_post_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            raise APIError(505, "error", "HTTP Version Not Supported")
+
+        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, 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
index 9a413940..6f2b350a 100644
--- a/segment/analytics/test/test_request.py
+++ b/segment/analytics/test/test_request.py
@@ -1,11 +1,12 @@
+import base64
 import json
 import unittest
-from datetime import date, datetime
+from datetime import date, datetime, timedelta, timezone
 from unittest import mock
 
 import requests
 
-from segment.analytics.request import DatetimeSerializer, post
+from segment.analytics.request import APIError, DatetimeSerializer, parse_retry_after, post
 
 
 class TestRequests(unittest.TestCase):
@@ -55,3 +56,165 @@ def mock_post_fn(*args, **kwargs):
             args, kwargs = mock_post.call_args
             self.assertIn("proxies", kwargs)
             self.assertEqual(kwargs["proxies"], proxies)
+
+    def test_authorization_header_basic_auth(self):
+        """Test that Basic Authorization header is added when no OAuth manager"""
+
+        def mock_post_fn(*args, **kwargs):
+            res = mock.Mock()
+            res.status_code = 200
+            return res
+
+        with mock.patch("segment.analytics.request._session.post", side_effect=mock_post_fn) as mock_post:
+            post("testsecret", batch=[{"userId": "userId", "event": "python event", "type": "track"}])
+
+            args, kwargs = mock_post.call_args
+            headers = kwargs["headers"]
+            self.assertIn("Authorization", headers)
+
+            # Verify it's Basic auth with correct encoding
+            expected_credentials = base64.b64encode(b"testsecret:").decode("utf-8")
+            expected_auth = f"Basic {expected_credentials}"
+            self.assertEqual(headers["Authorization"], expected_auth)
+
+    def test_authorization_header_oauth(self):
+        """Test that Bearer Authorization header is used with OAuth manager"""
+        oauth_manager = mock.Mock()
+        oauth_manager.get_token.return_value = "test_token_123"
+
+        def mock_post_fn(*args, **kwargs):
+            res = mock.Mock()
+            res.status_code = 200
+            return res
+
+        with mock.patch("segment.analytics.request._session.post", side_effect=mock_post_fn) as mock_post:
+            post("testsecret", oauth_manager=oauth_manager, batch=[{"userId": "userId", "event": "python event", "type": "track"}])
+
+            args, kwargs = mock_post.call_args
+            headers = kwargs["headers"]
+            self.assertIn("Authorization", headers)
+            self.assertEqual(headers["Authorization"], "Bearer test_token_123")
+
+    def test_x_retry_count_header(self):
+        """Test that X-Retry-Count header is omitted on first attempt and included on retries"""
+
+        def mock_post_fn(*args, **kwargs):
+            res = mock.Mock()
+            res.status_code = 200
+            return res
+
+        with mock.patch("segment.analytics.request._session.post", side_effect=mock_post_fn) as mock_post:
+            # Test with retry_count=0 (first attempt) — header should be absent
+            post("testsecret", retry_count=0, batch=[{"userId": "userId", "event": "python event", "type": "track"}])
+
+            args, kwargs = mock_post.call_args
+            headers = kwargs["headers"]
+            self.assertNotIn("X-Retry-Count", headers)
+
+        with mock.patch("segment.analytics.request._session.post", side_effect=mock_post_fn) as mock_post:
+            # Test with retry_count=5
+            post("testsecret", retry_count=5, batch=[{"userId": "userId", "event": "python event", "type": "track"}])
+
+            args, kwargs = mock_post.call_args
+            headers = kwargs["headers"]
+            self.assertEqual(headers["X-Retry-Count"], "5")
+
+    def test_non_200_2xx_treated_as_success(self):
+        """Test that 2xx and 3xx status codes are treated as success"""
+        for status_code in [200, 201, 204, 301, 302]:
+
+            def mock_post_fn(*args, **kwargs):
+                res = mock.Mock()
+                res.status_code = status_code
+                return res
+
+            with mock.patch("segment.analytics.request._session.post", side_effect=mock_post_fn):
+                res = post("testsecret", batch=[{"userId": "userId", "event": "python event", "type": "track"}])
+                self.assertEqual(res.status_code, status_code)
+
+    def test_parse_retry_after_integer(self):
+        """Test parsing Retry-After header with integer seconds"""
+        response = mock.Mock()
+        response.headers = {"Retry-After": "30"}
+        result = parse_retry_after(response)
+        self.assertEqual(result, 30)
+
+    def test_parse_retry_after_capped(self):
+        """Test that Retry-After is capped at 300 seconds"""
+        response = mock.Mock()
+        response.headers = {"Retry-After": "600"}
+        result = parse_retry_after(response)
+        self.assertEqual(result, 300)
+
+    def test_parse_retry_after_missing(self):
+        """Test parsing when Retry-After header is missing"""
+        response = mock.Mock()
+        response.headers = {}
+        result = parse_retry_after(response)
+        self.assertIsNone(result)
+
+    def test_parse_retry_after_invalid(self):
+        """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()
+        oauth_manager.get_token.return_value = "test_token"
+
+        def mock_post_fn(*args, **kwargs):
+            res = mock.Mock()
+            res.status_code = 511
+            res.json.return_value = {"code": "error", "message": "Network Authentication Required"}
+            return res
+
+        with mock.patch("segment.analytics.request._session.post", side_effect=mock_post_fn):
+            with self.assertRaises(APIError):
+                post("testsecret", oauth_manager=oauth_manager, batch=[{"userId": "userId", "event": "python event", "type": "track"}])
+
+            # Verify clear_token was called
+            oauth_manager.clear_token.assert_called_once()
+
+    def test_api_error_includes_response(self):
+        """Test that APIError includes the response object"""
+
+        def mock_post_fn(*args, **kwargs):
+            res = mock.Mock()
+            res.status_code = 429
+            res.json.return_value = {"code": "rate_limit", "message": "Too Many Requests"}
+            return res
+
+        with mock.patch("segment.analytics.request._session.post", side_effect=mock_post_fn):
+            try:
+                post("testsecret", batch=[{"userId": "userId", "event": "python event", "type": "track"}])
+            except APIError as e:
+                self.assertEqual(e.status, 429)
+                self.assertIsNotNone(e.response)
+            else:
+                self.fail("Expected APIError to be raised")
diff --git a/setup.py b/setup.py
index efd6fba3..6f35dc84 100644
--- a/setup.py
+++ b/setup.py
@@ -19,7 +19,7 @@
 Documentation and more details at https://github.com/segmentio/analytics-python
 """
 
-install_requires = ["requests~=2.7", "backoff~=2.1", "python-dateutil~=2.2", "PyJWT~=2.12"]
+install_requires = ["requests~=2.7", "python-dateutil~=2.2", "PyJWT~=2.12"]
 
 tests_require = [
     "mock==2.0.0",
diff --git a/uv.lock b/uv.lock
index 7e5db0be..34fdaafb 100644
--- a/uv.lock
+++ b/uv.lock
@@ -7,15 +7,6 @@ resolution-markers = [
     "python_full_version <= '3.9'",
 ]
 
-[[package]]
-name = "backoff"
-version = "2.2.1"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/47/d7/5bbeb12c44d7c4f2fb5b56abce497eb5ed9f34d85701de869acedd602619/backoff-2.2.1.tar.gz", hash = "sha256:03f829f5bb1923180821643f8753b0502c3b682293992485b0eef2807afa5cba", size = 17001, upload-time = "2022-10-05T19:19:32.061Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/df/73/b6e24bd22e6720ca8ee9a85a0c4a2971af8497d8f3193fa05390cbd46e09/backoff-2.2.1-py3-none-any.whl", hash = "sha256:63579f9a0628e06278f7e47b7d7d5b6ce20dc65c5e96a6f3ca99a6adca0396e8", size = 15148, upload-time = "2022-10-05T19:19:30.546Z" },
-]
-
 [[package]]
 name = "certifi"
 version = "2026.6.17"
@@ -717,7 +708,6 @@ name = "segment-analytics-python"
 version = "2.3.6"
 source = { editable = "." }
 dependencies = [
-    { name = "backoff" },
     { name = "pyjwt", extra = ["crypto"] },
     { name = "python-dateutil" },
     { name = "requests", version = "2.32.5", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" },
@@ -734,7 +724,6 @@ dev = [
 
 [package.metadata]
 requires-dist = [
-    { name = "backoff", specifier = "~=2.1" },
     { name = "pyjwt", extras = ["crypto"], specifier = "~=2.12" },
     { name = "python-dateutil", specifier = "~=2.2" },
     { name = "requests", specifier = "~=2.7" },