Dual-path retry: exponential backoff + rate-limit handling

HISTORY.md

@@ -1,3 +1,27 @@
+Unreleased
+==================
+
+### Upgrade note: new request header and proxy allowlists
+
+This release sends an `X-Retry-Count` request header on retries. If your
+traffic to Segment goes through a proxy, gateway or WAF that allowlists
+request headers, add it before upgrading or retried uploads will be
+rejected. The `Authorization` header is unchanged: this client has always
+sent the write key as HTTP Basic credentials.
+
+  * Send `X-Retry-Count` on retries from both the LibCurl and Socket consumers, 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, 505 and 511) are retried. `Retry-After` is honoured on all of them, not just 429, which brings 529 in through the generic 5xx rule.
+  * `Retry-After` accepts numeric seconds and the RFC 7231 HTTP-date formats, capped at 300s (`rate_limit_retry_after_cap`). Malformed values are rejected rather than parsed into an arbitrary date.
+  * 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 options `max_total_backoff_duration` and `max_rate_limit_duration`, both in seconds and defaulting to 12 hours, bound the two waits.
+  * The Socket consumer now honours `retry_count` and `max_total_backoff_duration` too, and backs off from 500ms like the LibCurl consumer rather than 100ms. It previously ignored both and gave up after a fixed seven retries over roughly 13 seconds. `maximum_backoff_duration` now caps each individual wait rather than ending the loop, so its 10s default still bounds how long any one retry sleeps. `max_rate_limit_duration` does not apply there, since Socket still does not read `Retry-After` — use the LibCurl consumer if you need that.
+  * `retry_count` grants exactly that many retries. It previously granted one fewer, and a `retry_count` of 1 granted none.
+  * The new budget options reject negative values and keep the default, logging as `flush_at` and `flush_interval` already do. A negative previously disabled retrying outright. Zero is still accepted and meaningful: `retry_count` of 0 means do not retry.
+  * Transport failures report the real libcurl error number to `error_handler` again, so a DNS failure, a timeout and a TLS error can be told apart.
+  * 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 curl already declined to follow will not succeed on a retry. The Segment endpoint does not redirect, so this only affects custom `host` values.
+  * Retry timing uses `hrtime()`, so a system clock change cannot stretch or collapse a backoff.
+  * Fix an oversized batch wedging the queue: the batch is now removed before the size check, so one too-large batch no longer makes every later `track()` return false.
+
 
 3.8.2 / 2026-03-11
 ==================

e2e-cli/e2e-config.json

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

e2e-cli/main.php

@@ -135,10 +135,9 @@ function parseHost(string $apiHost): string
  * Build the options array for Segment\Client.
  *
  * @param array<string,mixed>   $input
- * @param array<int,string>    &$errors   collected error messages
  * @return array<string,mixed>
  */
-function buildClientOptions(array $input, array &$errors): array
+function buildClientOptions(array $input): array
 {
     $config  = $input['config'] ?? [];
     $apiHost = $input['apiHost'] ?? '';
@@ -154,10 +153,11 @@ function buildClientOptions(array $input, array &$errors): array
         // mock test server (the base LibCurl hardcodes https://).
         'consumer'      => E2eLibCurl::class,
         'protocol'      => $scheme,
-        'error_handler' => function (int $code, string $message) use (&$errors): void {
-            $msg = "HTTP {$code}: {$message}";
-            debugLog('SDK error — ' . $msg);
-            $errors[] = $msg;
+        // Log HTTP errors to stderr only — success/failure is determined by
+        // track()/flush() return values, not by the error_handler callback,
+        // because handleError fires for transient retry errors too.
+        'error_handler' => function (int $code, string $message): void {
+            debugLog("SDK HTTP error {$code}: {$message}");
         },
     ];
 
@@ -176,6 +176,11 @@ function buildClientOptions(array $input, array &$errors): array
         debugLog('curl_timeout: ' . $options['curl_timeout']);
     }
 
+    if (isset($config['maxRetries']) && is_numeric($config['maxRetries'])) {
+        $options['retry_count'] = (int)$config['maxRetries'];
+        debugLog('retry_count: ' . $options['retry_count']);
+    }
+
     return $options;
 }
 
@@ -241,9 +246,10 @@ function buildMessage(array $event): array
 }
 
 $errors = [];
+$autoFlushFailed = false; // set true if an enqueue() auto-flush returns false
 
-// Build client options (error_handler captures into $errors by reference)
-$options = buildClientOptions($input, $errors);
+// Build client options (error_handler just logs; we track success via return values)
+$options = buildClientOptions($input);
 
 debugLog('Creating Segment\\Client with writeKey=' . substr($writeKey, 0, 4) . '...');
 
@@ -268,30 +274,35 @@ function buildMessage(array $event): array
 
         debugLog("  [{$seqIndex}/{$eventIndex}] Enqueueing {$type}");
 
+        $enqueueOk = true;
         switch ($type) {
             case 'track':
-                $client->track($message);
+                $enqueueOk = $client->track($message);
                 break;
             case 'identify':
-                $client->identify($message);
+                $enqueueOk = $client->identify($message);
                 break;
             case 'page':
-                $client->page($message);
+                $enqueueOk = $client->page($message);
                 break;
             case 'screen':
-                $client->screen($message);
+                $enqueueOk = $client->screen($message);
                 break;
             case 'alias':
-                $client->alias($message);
+                $enqueueOk = $client->alias($message);
                 break;
             case 'group':
-                $client->group($message);
+                $enqueueOk = $client->group($message);
                 break;
             default:
                 $errors[] = "Unknown event type: {$type}";
                 debugLog("  Unknown event type: {$type}");
                 break;
         }
+        if (!$enqueueOk) {
+            $autoFlushFailed = true;
+            debugLog("  Enqueue/auto-flush failed for {$type}");
+        }
     }
 }
 
@@ -306,14 +317,19 @@ function buildMessage(array $event): array
     $errors[] = 'Flush failed';
 }
 
-$hasErrors = !empty($errors);
-$success   = $flushOk && !$hasErrors;
+// Success = all flushes succeeded and no fatal errors.
+// auto-flushes (from enqueue when flush_at reached) and explicit flush are both tracked.
+$overallSuccess = $flushOk && !$autoFlushFailed && empty($errors);
 
-if ($success) {
+if ($overallSuccess) {
     outputResult(true, $sentBatches);
     exit(0);
 } else {
-    $errorMsg = implode('; ', $errors);
+    $allErrors = array_merge(
+        $errors,
+        $autoFlushFailed ? ['Auto-flush failed'] : []
+    );
+    $errorMsg = implode('; ', $allErrors ?: ['Unknown flush failure']);
     outputResult(false, $sentBatches, $errorMsg);
     exit(1);
 }

lib/Consumer/LibCurl.php

@@ -9,92 +9,171 @@ class LibCurl extends QueueConsumer
     protected string $type = 'LibCurl';
 
     /**
-     * Make a sync request to our API. If debug is
-     * enabled, we wait for the response
-     * and retry once to diminish impact on performance.
+     * Send a batch of messages to the API with retries on error
+     *
      * @param array $messages array of all the messages to send
      * @return bool whether the request succeeded
      */
     public function flushBatch(array $messages): bool
     {
-        $body = $this->payload($messages);
+        $body    = $this->payload($messages);
         $payload = json_encode($body);
-        $secret = $this->secret;
+        $secret  = $this->secret;
 
         if ($this->compress_request) {
             $payload = gzencode($payload);
         }
 
-        if ($this->host) {
-            $host = $this->host;
-        } else {
-            $host = 'api.segment.io';
-        }
-        $path = '/v1/batch';
-        $url = $this->protocol . $host . $path;
+        $host = $this->host ?: 'api.segment.io';
+        $url  = $this->protocol . $host . '/v1/batch';
 
-        $backoff = 100; // Set initial waiting time to 100ms
+        $library   = $messages[0]['context']['library'];
+        $userAgent = $library['name'] . '/' . $library['version'];
 
-        while ($backoff < $this->maximum_backoff_duration) {
-            // open connection
-            $ch = curl_init();
+        $backoffMs          = 500;   // base 500ms per spec
+        $backoffCapMs       = 60000; // cap 60s
+        $retriesRemaining   = $this->retry_count;
+        $attempt            = 0;
+        $backoffStartTime   = null;
+        $rateLimitStartTime = null;
 
-            // set the url, number of POST vars, POST data
-            curl_setopt($ch, CURLOPT_USERPWD, $secret . ':');
-            curl_setopt($ch, CURLOPT_POSTFIELDS, $payload);
-            curl_setopt($ch, CURLOPT_TIMEOUT, $this->curl_timeout);
-            curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, $this->curl_connecttimeout);
+        while (true) {
+            $attempt++;
 
-            // set variables for headers
-            $header = [];
-            $header[] = 'Content-Type: application/json';
+            $headers = [
+                'Content-Type: application/json',
+                'User-Agent: ' . $userAgent,
+            ];
 
             if ($this->compress_request) {
-                $header[] = 'Content-Encoding: gzip';
+                $headers[] = 'Content-Encoding: gzip';
             }
 
-            // Send user agent in the form of {library_name}/{library_version} as per RFC 7231.
-            $library = $messages[0]['context']['library'];
-            $libName = $library['name'];
-            $libVersion = $library['version'];
-            $header[] = "User-Agent: $libName/$libVersion";
-
-            curl_setopt($ch, CURLOPT_HTTPHEADER, $header);
-            curl_setopt($ch, CURLOPT_URL, $url);
-            curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
+            if ($attempt > 1) {
+                $headers[] = 'X-Retry-Count: ' . ($attempt - 1);
+            }
 
-            // retry failed requests just once to diminish impact on performance
-            $responseContent = curl_exec($ch);
+            [$responseCode, $responseHeaders, $responseContent, $err, $errno] =
+                $this->executeHttpRequest($url, $secret, $payload, $headers);
 
-            $err = curl_error($ch);
             if ($err) {
-                $this->handleError(curl_errno($ch), $err);
+                // The real libcurl errno, not 0: error_handler callbacks branch on it
+                // to tell a DNS failure from a timeout from a TLS error.
+                $this->handleError($errno, $err);
+
                 return false;
             }
 
-            $responseCode = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
+            // Only 2xx is success. curl is not configured to follow redirects, so a
+            // 3xx means nothing was uploaded; treating it as success would drop the
+            // batch silently. TAPI does not emit 3xx — this shows up when the
+            // configured host is a proxy or redirector.
+            if ($responseCode >= 200 && $responseCode < 300) {
+                return true;
+            }
 
-            //close connection
-            curl_close($ch);
+            if ($responseCode >= 300 && $responseCode < 400) {
+                $this->handleError(
+                    $responseCode,
+                    'Unexpected redirect; batch not uploaded. Check whether the configured '
+                    . 'host points at a proxy or redirector.'
+                );
+                return false;
+            }
 
-            if ($responseCode !== 200) {
-                // log error
-                $this->handleError($responseCode, $responseContent);
+            $this->handleError($responseCode, $responseContent);
 
-                if (($responseCode >= 500 && $responseCode <= 600) || $responseCode === 429) {
-                    // If status code is greater than 500 and less than 600, it indicates server error
-                    // Error code 429 indicates rate limited.
-                    // Retry uploading in these cases.
-                    usleep($backoff * 1000);
-                    $backoff *= 2;
-                } elseif ($responseCode >= 400) {
-                    break;
+            if (!$this->isRetryable($responseCode)) {
+                return false;
+            }
+
+            // Any retryable status with valid Retry-After: use rate-limit path (no budget cost)
+            $retryAfterS = $this->parseRetryAfter($responseHeaders['retry-after'] ?? null);
+            if ($retryAfterS !== null) {
+                if ($rateLimitStartTime === null) {
+                    // hrtime is monotonic; microtime would let a clock adjustment
+                    // expire or extend this budget.
+                    $rateLimitStartTime = hrtime(true);
                 }
-            } else {
-                break; // no error
+                if ((hrtime(true) - $rateLimitStartTime) / 1e6 >= $this->max_rate_limit_duration_ms) {
+                    return false;
+                }
+                $sleepMs = min($retryAfterS * 1000, $this->rate_limit_retry_after_cap_s * 1000);
+                $this->sleepBeforeRetry($sleepMs, true);
+                continue; // Do NOT decrement retriesRemaining
+            }
+
+            // No Retry-After: counted backoff
+            // Checked before the decrement: decrementing first spent one retry on
+            // the exhaustion test itself, so retry_count of N performed N-1 and a
+            // retry_count of 1 performed none at all.
+            if ($retriesRemaining <= 0) {
+                return false;
+            }
+            $retriesRemaining--;
+            if ($backoffStartTime === null) {
+                $backoffStartTime = hrtime(true);
             }
+            if ((hrtime(true) - $backoffStartTime) / 1e6 >= $this->max_total_backoff_duration_ms) {
+                return false;
+            }
+            $this->sleepBeforeRetry($backoffMs, false);
+            $backoffMs = min($backoffMs * 2, $backoffCapMs);
         }
+    }
+
+    /**
+     * Wait before the next attempt. Separate from flushBatch so tests can observe the
+     * retry schedule by overriding this alone.
+     *
+     * @param int  $milliseconds how long to wait
+     * @param bool $rateLimited  true when the server sent Retry-After, false for counted backoff
+     */
+    protected function sleepBeforeRetry(int $milliseconds, bool $rateLimited): void
+    {
+        usleep($milliseconds * 1000);
+    }
+
+    /**
+     * Execute an HTTP POST request via cURL.
+     *
+     * Returns [statusCode, responseHeaders, responseBody, curlError, curlErrno].
+     * responseHeaders keys are lower-cased.
+     *
+     * @param string $url
+     * @param string $secret
+     * @param string $payload
+     * @param array  $headers
+     * @return array{int, array<string,string>, string|false, string, int}
+     */
+    protected function executeHttpRequest(string $url, string $secret, string $payload, array $headers): array
+    {
+        $responseHeaders = [];
+
+        $ch = curl_init();
+
+        curl_setopt($ch, CURLOPT_USERPWD, $secret . ':');
+        curl_setopt($ch, CURLOPT_POSTFIELDS, $payload);
+        curl_setopt($ch, CURLOPT_TIMEOUT, $this->curl_timeout);
+        curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, $this->curl_connecttimeout);
+        curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);
+        curl_setopt($ch, CURLOPT_URL, $url);
+        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
+        curl_setopt($ch, CURLOPT_HEADERFUNCTION, function ($curl, $header) use (&$responseHeaders) {
+            $parts = explode(':', $header, 2);
+            if (count($parts) === 2) {
+                $responseHeaders[strtolower(trim($parts[0]))] = trim($parts[1]);
+            }
+
+            return strlen($header);
+        });
+
+        $responseContent = curl_exec($ch);
+        $err             = curl_error($ch);
+        $errno           = curl_errno($ch);
+        $responseCode    = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
+        curl_close($ch);
 
-        return true;
+        return [$responseCode, $responseHeaders, $responseContent, $err, $errno];
     }
 }