Improve HTTP response handling and retry logic

HISTORY.md

@@ -1,3 +1,22 @@
+# Unreleased
+### Upgrade note: new request headers and proxy allowlists
+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
+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.
+
 # 2.3.6 / 2026-4-7
 - Update and widen PyJWT version to address security issue
 

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

e2e-cli/e2e-config.json

@@ -1,7 +1,9 @@
 {
   "sdk": "python",
-  "test_suites": "basic",
+  "test_suites": "basic,retry",
   "auto_settings": false,
   "patch": null,
-  "env": {}
+  "env": {
+    "AUTH_HEADER": "true"
+  }
 }

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

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

pyproject.toml

@@ -26,7 +26,6 @@ classifiers = [
 ]
 dependencies = [
     "requests~=2.7",
-    "backoff~=2.1",
     "python-dateutil~=2.2",
     "PyJWT[crypto]~=2.12",
 ]

requirements.txt

@@ -1,4 +1,3 @@
-backoff==2.2.1
 cryptography==44.0.0
 flake8==7.1.1
 mock==2.0.0

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