fix: handle retryable http responses across consumers

.changeset/consumer-http-retries.md

@@ -0,0 +1,5 @@
+---
+"posthog-php": patch
+---
+
+Retry transient HTTP responses and honor Retry-After in Socket and foreground ForkCurl consumers. Retry-After waits are now capped at 10 seconds in all HTTP transports. When retries run out on a retryable response, all HTTP consumers now keep the batch queued instead of dropping it, and automatic flushes wait for the maximum backoff duration before they try again.

api/public-api.json

@@ -3344,6 +3344,17 @@
                 }
             }
         },
+        "PostHog\\HttpRetryPolicy": {
+            "type": "trait",
+            "abstract": false,
+            "final": false,
+            "readonly": false,
+            "extends": null,
+            "implements": [],
+            "constants": [],
+            "properties": [],
+            "methods": []
+        },
         "PostHog\\InconclusiveMatchException": {
             "type": "class",
             "abstract": false,

lib/Consumer/ForkCurl.php

@@ -3,6 +3,7 @@
 namespace PostHog\Consumer;
 
 use PostHog\QueueConsumer;
+use Symfony\Component\Clock\Clock;
 
 /**
  * Queue consumer that sends batches through a forked curl process.
@@ -93,6 +94,16 @@ public function flushBatch($messages)
         // Send user agent in the form of {library_name}/{library_version} as per RFC 7231.
         $cmd .= " -H 'User-Agent: " . $this->userAgent() . "'";
 
+        if ($this->debug()) {
+            try {
+                return $this->sendForeground($cmd);
+            } finally {
+                if ($tmpfname != "") {
+                    unlink($tmpfname);
+                }
+            }
+        }
+
         if (!$this->debug()) {
             $cmd .= " > /dev/null 2>&1 &";
         }
@@ -116,6 +127,50 @@ public function flushBatch($messages)
             : self::FLUSH_BATCH_NON_RETRYABLE_FAILURE;
     }
 
+    private function sendForeground(string $cmd): bool|string
+    {
+        $headersFile = tempnam(sys_get_temp_dir(), "posthog-curl-headers-");
+        if ($headersFile === false) {
+            return self::FLUSH_BATCH_NON_RETRYABLE_FAILURE;
+        }
+
+        try {
+            $cmd .= " --silent --show-error --dump-header " . escapeshellarg($headersFile);
+            $cmd .= " --write-out '\n%{http_code}'";
+            $backoff = 100;
+
+            while (true) {
+                $output = [];
+                exec($cmd, $output, $exit);
+                if ($exit !== 0) {
+                    $this->handleError($exit, implode("\n", $output));
+                    return $this->isNetworkCurlExit($exit)
+                        ? self::FLUSH_BATCH_RETRYABLE_FAILURE
+                        : self::FLUSH_BATCH_NON_RETRYABLE_FAILURE;
+                }
+
+                $statusCode = (int) array_pop($output);
+                if ($statusCode === 200) {
+                    return true;
+                }
+
+                $this->handleError($statusCode, implode("\n", $output));
+                if (!$this->isRetryableStatus($statusCode)) {
+                    return self::FLUSH_BATCH_NON_RETRYABLE_FAILURE;
+                }
+                if ($backoff >= $this->maximum_backoff_duration) {
+                    return self::FLUSH_BATCH_RETRYABLE_FAILURE;
+                }
+
+                $headers = file($headersFile, FILE_IGNORE_NEW_LINES) ?: [];
+                Clock::get()->sleep($this->retryDelayMilliseconds($headers, $backoff) / 1000);
+                $backoff *= 2;
+            }
+        } finally {
+            unlink($headersFile);
+        }
+    }
+
     private function isNetworkCurlExit($exit)
     {
         return in_array((int) $exit, [6, 7, 28, 35, 52, 56], true);

lib/Consumer/LibCurl.php

@@ -106,7 +106,7 @@ public function flushBatch($messages)
             return true;
         }
 
-        return $this->isNetworkFailure($response)
+        return $this->isNetworkFailure($response) || $this->isRetryableStatus($response->getResponseCode())
             ? self::FLUSH_BATCH_RETRYABLE_FAILURE
             : self::FLUSH_BATCH_NON_RETRYABLE_FAILURE;
     }

lib/Consumer/Socket.php

@@ -4,6 +4,7 @@
 
 use Exception;
 use PostHog\QueueConsumer;
+use Symfony\Component\Clock\Clock;
 
 /**
  * Queue consumer that sends batches over a raw socket.
@@ -60,9 +61,7 @@ public function flushBatch($batch)
 
         $body = $this->createBody($this->host, $payload);
 
-        return $this->makeRequest($socket, $body)
-            ? true
-            : self::FLUSH_BATCH_NON_RETRYABLE_FAILURE;
+        return $this->makeRequest($socket, $body);
     }
 
     private function createSocket()
@@ -103,13 +102,47 @@ private function createSocket()
         }
     }
 
+    private function readResponse($socket): string|false
+    {
+        $maximumHeaderBytes = 64 * 1024;
+        $timeout = max(0.000001, (float) $this->options['timeout']);
+        $defaultTimeout = (float) ini_get('default_socket_timeout');
+        if ($defaultTimeout > 0) {
+            $timeout = min($timeout, $defaultTimeout);
+        }
+        $deadline = hrtime(true) / 1e9 + $timeout;
+        $response = '';
+
+        while (true) {
+            $headerEnd = strpos($response, "\r\n\r\n");
+            if ($headerEnd !== false) {
+                return $headerEnd + 4 <= $maximumHeaderBytes ? $response : false;
+            }
+            if (strlen($response) >= $maximumHeaderBytes || feof($socket)) {
+                return false;
+            }
+
+            $remaining = $deadline - hrtime(true) / 1e9;
+            if ($remaining <= 0) {
+                return false;
+            }
+            $microseconds = (int) ceil($remaining * 1000000);
+            stream_set_timeout($socket, intdiv($microseconds, 1000000), $microseconds % 1000000);
+            $chunk = fread($socket, min(2048, $maximumHeaderBytes - strlen($response)));
+            if ($chunk === false || $chunk === '' || stream_get_meta_data($socket)['timed_out']) {
+                return false;
+            }
+            $response .= $chunk;
+        }
+    }
+
     /**
      * Attempt to write the request to the socket, wait for response if debug
      * mode is enabled.
      * @param resource $socket the handle for the socket
      * @param string $req request body
      * @param boolean $retry
-     * @return boolean $success
+     * @return bool|string True on success, or a queue failure classification.
      */
     private function makeRequest($socket, $req, $retry = true)
     {
@@ -141,9 +174,11 @@ private function makeRequest($socket, $req, $retry = true)
             // Get response for request
             $statusCode = 0;
             $errorMessage = "";
+            $response = "";
 
             if (!$closed) {
-                $res = $this->parseResponse(fread($socket, 2048));
+                $response = $this->readResponse($socket) ?: '';
+                $res = $this->parseResponse($response);
                 $statusCode = (int)$res["status"];
                 $errorMessage = $res["message"];
             }
@@ -157,12 +192,13 @@ private function makeRequest($socket, $req, $retry = true)
             // 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.
-            if (($statusCode >= 500 && $statusCode <= 600) || 429 == $statusCode || 0 == $statusCode) {
+            if ($this->isRetryableStatus($statusCode) || 0 == $statusCode) {
                 if ($backoff >= $this->maximum_backoff_duration) {
-                    break;
+                    return self::FLUSH_BATCH_RETRYABLE_FAILURE;
                 }
 
-                usleep($backoff * 1000);
+                $responseHeaders = explode("\r\n", strstr($response, "\r\n\r\n", true) ?: '');
+                Clock::get()->sleep($this->retryDelayMilliseconds($responseHeaders, $backoff) / 1000);
             } elseif ($statusCode >= 400) {
                 if ($this->debug()) {
                     $this->handleError($res["status"], $res["message"]);
@@ -179,7 +215,7 @@ private function makeRequest($socket, $req, $retry = true)
             }
         }
 
-        return false;
+        return self::FLUSH_BATCH_NON_RETRYABLE_FAILURE;
     }
 
     /**

lib/HttpClient.php

@@ -3,6 +3,7 @@
 namespace PostHog;
 
 use Closure;
+use Symfony\Component\Clock\Clock;
 
 /**
  * HTTP client used by the SDK for PostHog API requests.
@@ -11,6 +12,8 @@
  */
 class HttpClient
 {
+    use HttpRetryPolicy;
+
     /**
      * @var string
      */
@@ -164,8 +167,7 @@ public function sendRequest(string $path, ?string $payload, array $extraHeaders
                 } elseif ($this->isRetryableStatus($responseCode)) {
                     // Retry transient failures. Honor Retry-After when provided, otherwise use
                     // exponential backoff starting at 100ms.
-                    $retryAfterMs = $this->retryAfterMilliseconds($responseHeaders);
-                    usleep(($retryAfterMs ?? $backoff) * 1000);
+                    Clock::get()->sleep($this->retryDelayMilliseconds($responseHeaders, $backoff) / 1000);
                     $backoff *= 2;
                 } else {
                     // Do not retry non-transient responses (e.g. 400, 401, 403, 413,
@@ -205,41 +207,6 @@ private function executePost($ch, bool $includeEtag = false): HttpResponse
         return new HttpResponse($response, $responseCode, null, $curlErrno);
     }
 
-    protected function isRetryableStatus(int $responseCode): bool
-    {
-        return $responseCode === 408
-            || $responseCode === 429
-            || ($responseCode >= 500 && $responseCode <= 600);
-    }
-
-    /**
-     * @param array<int, string> $headers
-     */
-    protected function retryAfterMilliseconds(array $headers): ?int
-    {
-        foreach ($headers as $header) {
-            if (stripos($header, 'Retry-After:') !== 0) {
-                continue;
-            }
-
-            $value = trim(substr($header, strlen('Retry-After:')));
-            if ($value === '') {
-                return null;
-            }
-
-            if (ctype_digit($value)) {
-                return max(0, (int) $value * 1000);
-            }
-
-            $timestamp = strtotime($value);
-            if ($timestamp !== false) {
-                return max(0, (int) (($timestamp - time()) * 1000));
-            }
-        }
-
-        return null;
-    }
-
     private function handleError($code, $message)
     {
         if (null !== $this->errorHandler) {

lib/HttpRetryPolicy.php

@@ -0,0 +1,54 @@
+<?php
+
+namespace PostHog;
+
+/** @internal */
+trait HttpRetryPolicy
+{
+    private const MAXIMUM_RETRY_AFTER_MILLISECONDS = 10000;
+
+    protected function isRetryableStatus(int $responseCode): bool
+    {
+        return $responseCode === 408
+            || $responseCode === 429
+            || ($responseCode >= 500 && $responseCode <= 600);
+    }
+
+    /**
+     * @param array<int, string> $headers
+     */
+    protected function retryDelayMilliseconds(array $headers, int $backoff): int
+    {
+        $retryAfter = $this->retryAfterMilliseconds($headers);
+
+        return $retryAfter === null ? $backoff : min($retryAfter, self::MAXIMUM_RETRY_AFTER_MILLISECONDS);
+    }
+
+    /**
+     * @param array<int, string> $headers
+     */
+    protected function retryAfterMilliseconds(array $headers): ?int
+    {
+        foreach ($headers as $header) {
+            if (stripos($header, 'Retry-After:') !== 0) {
+                continue;
+            }
+
+            $value = trim(substr($header, strlen('Retry-After:')));
+            if ($value === '') {
+                return null;
+            }
+
+            if (ctype_digit($value)) {
+                return min((int) $value, intdiv(PHP_INT_MAX, 1000)) * 1000;
+            }
+
+            $timestamp = strtotime($value);
+            if ($timestamp !== false) {
+                return max(0, (int) (($timestamp - time()) * 1000));
+            }
+        }
+
+        return null;
+    }
+}

lib/QueueConsumer.php

@@ -15,6 +15,8 @@
  */
 abstract class QueueConsumer extends Consumer
 {
+    use HttpRetryPolicy;
+
     protected const MAX_BATCH_PAYLOAD_SIZE = 1024 * 1024; // 1MB
     protected const MAX_BATCH_PAYLOAD_SIZE_HUMAN = (self::MAX_BATCH_PAYLOAD_SIZE / 1024) . 'KB';
 
@@ -31,6 +33,7 @@ abstract class QueueConsumer extends Consumer
     protected $batch_size = 100;
     protected $flush_interval = 5.0;
     protected $flush_after = null;
+    protected $retry_cooldown_until = null;
     protected $maximum_backoff_duration = 10000;    // Set maximum waiting limit to 10s
     protected $host = "us.i.posthog.com";
     protected $compress_request = false;
@@ -142,6 +145,10 @@ public function flush()
             $result = $this->flushBatch($batch);
             $success = true === $result;
 
+            if (self::FLUSH_BATCH_RETRYABLE_FAILURE === $result) {
+                $this->retry_cooldown_until = $this->now() + $this->maximum_backoff_duration / 1000;
+            }
+
             if ($success || self::FLUSH_BATCH_RETRYABLE_FAILURE !== $result) {
                 array_splice($this->queue, 0, $batchSize);
             }
@@ -174,6 +181,10 @@ public function enqueue($item)
             $this->resetFlushTimer();
         }
 
+        if ($this->inRetryCooldown()) {
+            return true;
+        }
+
         if ($this->flush_interval === 0.0 || $count >= $this->batch_size || $this->flushIntervalElapsed()) {
             return $this->flush(); // return ->flush() result: true on success
         }
@@ -193,6 +204,11 @@ private function flushIntervalElapsed(): bool
         return $this->flush_after !== null && $this->now() >= $this->flush_after;
     }
 
+    private function inRetryCooldown(): bool
+    {
+        return $this->retry_cooldown_until !== null && $this->now() < $this->retry_cooldown_until;
+    }
+
     private function now(): float
     {
         return (float) Clock::get()->now()->format('U.u');