diff --git a/.changeset/consumer-http-retries.md b/.changeset/consumer-http-retries.md new file mode 100644 index 0000000..7c729bc --- /dev/null +++ b/.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. diff --git a/api/public-api.json b/api/public-api.json index d164eae..b0c4e81 100644 --- a/api/public-api.json +++ b/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, diff --git a/lib/Consumer/ForkCurl.php b/lib/Consumer/ForkCurl.php index 9d4f8e3..b7ab9ba 100644 --- a/lib/Consumer/ForkCurl.php +++ b/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); diff --git a/lib/Consumer/LibCurl.php b/lib/Consumer/LibCurl.php index 24fd727..0a80c5a 100644 --- a/lib/Consumer/LibCurl.php +++ b/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; } diff --git a/lib/Consumer/Socket.php b/lib/Consumer/Socket.php index f65970b..d7a7fcd 100644 --- a/lib/Consumer/Socket.php +++ b/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; } /** diff --git a/lib/HttpClient.php b/lib/HttpClient.php index 45583f3..7a5d752 100644 --- a/lib/HttpClient.php +++ b/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 $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) { diff --git a/lib/HttpRetryPolicy.php b/lib/HttpRetryPolicy.php new file mode 100644 index 0000000..4bffd36 --- /dev/null +++ b/lib/HttpRetryPolicy.php @@ -0,0 +1,54 @@ += 500 && $responseCode <= 600); + } + + /** + * @param array $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 $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; + } +} diff --git a/lib/QueueConsumer.php b/lib/QueueConsumer.php index f28a814..878b93c 100644 --- a/lib/QueueConsumer.php +++ b/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'); diff --git a/sdk_compliance_adapter/README.md b/sdk_compliance_adapter/README.md index a5b2f1e..aebded7 100644 --- a/sdk_compliance_adapter/README.md +++ b/sdk_compliance_adapter/README.md @@ -29,8 +29,8 @@ uncompressed HTTP client even when capture compression is enabled. The adapter maps `flush_at` to `batch_size` and `flush_interval_ms` to `flush_interval_seconds` (SDK default: 5 seconds, enforced on enqueue, not an idle timer). `max_retries` maps to the existing backoff-duration option: for three -retries, 801 ms for HttpClient and 800 ms for Socket, reflecting their different -limit checks. ForkCurl does not implement that retry option. Socket's connection +retries, 801 ms for HttpClient and 800 ms for Socket and foreground ForkCurl, +reflecting their different limit checks. Socket's connection timeout is expressed in seconds; the PHP worker's `default_socket_timeout=1` bounds reads, including idle retry connections. This is a configured runtime profile, not certification of default PHP socket timing. @@ -66,31 +66,25 @@ Reset kills and waits for only the owned PHP worker, without running its queue-flushing destructors. It then closes the relay. This provides test isolation, not SDK shutdown certification; `/flush` always calls the real public SDK method. -## Known failures and applicability - -LibCurl is expected to pass all 47 selected tests. Foreground ForkCurl exposes -these 11 capture failures. The last Socket report also contains these failures; -it predates the Socket write-state repair: - -- `capture.retry_behavior.retries_on_503` -- `capture.retry_behavior.retries_on_500` -- `capture.retry_behavior.retries_on_502` -- `capture.retry_behavior.retries_on_504` -- `capture.retry_behavior.respects_retry_after_header` -- `capture.retry_behavior.implements_backoff` -- `capture.retry_behavior.max_retries_respected` -- `capture.error_handling.retries_on_408` -- `capture.deduplication.preserves_uuid_on_retry` -- `capture.deduplication.preserves_uuid_and_timestamp_on_retry` -- `capture.deduplication.preserves_uuid_and_timestamp_on_batch_retry` - -Socket now resends the complete request on each attempt. Native regression tests -cover 503 and 429 followed by 200, stable request bodies, and queue drain. The full -Socket compliance suite was not rerun for that repair. Socket still treats 408 as -terminal and ignores Retry-After. ForkCurl does not retry HTTP rejection and -reports a successful curl -process exit as successful delivery even for HTTP errors. The adapter exposes -these behaviors rather than retrying on the SDK's behalf. +## Retry policy and applicability + +Socket and foreground ForkCurl retry transient HTTP failures, including 408, +429, and server errors. Each attempt sends the same batch. Retry delays start at +100 ms and double after each attempt, within the configured backoff limit. +Valid Retry-After headers replace the delay for that attempt. These headers can +contain a number of seconds or an HTTP date. Invalid headers use the normal +backoff delay; past dates permit an immediate retry. + +HTTP 200 confirms delivery. Terminal HTTP errors make flush return false and +remove the failed batch. Retry exhaustion on a retryable response or a network +failure makes flush return false and keeps the batch queued. Automatic flushes +then wait for the maximum backoff duration before they try again; an explicit +flush tries at once. Foreground ForkCurl uses the system curl command without +requiring PHP's curl extension for batch delivery. + +Native transport tests cover status handling, retry limits, Retry-After parsing, +stable request bodies, gzip, and queue drain. The full compliance reports measure +each profile against the selected harness tests. Background ForkCurl and LibCurl's explicit fire-and-forget option do not promise blocking verified delivery and are not selected. File/noop consumers are not HTTP diff --git a/sdk_compliance_adapter/adapter.php b/sdk_compliance_adapter/adapter.php index b19a0dc..7ff0485 100644 --- a/sdk_compliance_adapter/adapter.php +++ b/sdk_compliance_adapter/adapter.php @@ -69,7 +69,7 @@ function maxBackoffDurationForRetries(int $maxRetries, string $consumer): int } // Socket checks its limit before sleeping; HttpClient checks after doubling. - return (100 * (2 ** $maxRetries)) + ($consumer === 'socket' ? 0 : 1); + return (100 * (2 ** $maxRetries)) + ($consumer === 'lib_curl' ? 1 : 0); } function handleRequest(array $request, AdapterState $state): array diff --git a/sdk_compliance_adapter/test_server.py b/sdk_compliance_adapter/test_server.py index 1d72501..8324af8 100644 --- a/sdk_compliance_adapter/test_server.py +++ b/sdk_compliance_adapter/test_server.py @@ -111,15 +111,23 @@ def test_compressed_capture_does_not_change_flags_headers_or_side_effects(self): self.assertEqual(state["total_retries"], 0) def test_production_retry_attempts_are_observed(self): - self.init() - self.mock.statuses = [503, 200] - captured = self.capture() - self.assertTrue(self.call("/flush", {})["success"]) - requests = self.call("/state")["requests_made"] - self.assertEqual([r["status_code"] for r in requests], [503, 200]) - self.assertEqual([r["retry_attempt"] for r in requests], [0, 1]) - self.assertTrue(all(r["uuid_list"] == [captured["uuid"]] for r in requests)) - self.assertEqual(self.mock.requests[0][2], self.mock.requests[1][2]) + cases = ( + ("successful retry", [503, 200], 3, True, [503, 200]), + ("retry limit", [503, 503, 200], 1, False, [503, 503]), + ("retries disabled", [503, 200], 0, False, [503]), + ) + for consumer in ("lib_curl", "socket", "fork_curl"): + for name, statuses, max_retries, success, expected_statuses in cases: + with self.subTest(consumer=consumer, case=name): + self.init(consumer, max_retries=max_retries) + self.mock.statuses = statuses.copy() + captured = self.capture() + self.assertEqual(self.call("/flush", {})["success"], success) + requests = self.call("/state")["requests_made"] + self.assertEqual([r["status_code"] for r in requests], expected_statuses) + self.assertEqual([r["retry_attempt"] for r in requests], list(range(len(expected_statuses)))) + self.assertTrue(all(r["uuid_list"] == [captured["uuid"]] for r in requests)) + self.assertTrue(all(r[2] == self.mock.requests[0][2] for r in self.mock.requests)) def test_terminal_response_exposes_sdk_boolean_not_observer_success(self): for consumer in ("lib_curl", "socket", "fork_curl"): @@ -128,7 +136,7 @@ def test_terminal_response_exposes_sdk_boolean_not_observer_success(self): self.mock.statuses = [400] self.capture() flushed = self.call("/flush", {}) - self.assertEqual(flushed["success"], consumer == "fork_curl") + self.assertFalse(flushed["success"]) self.assertEqual(flushed["events_flushed"], 0) self.assertEqual(self.call("/state")["requests_made"][0]["status_code"], 400) diff --git a/test/ConsumerForkCurlTest.php b/test/ConsumerForkCurlTest.php index 6657d60..29cccac 100644 --- a/test/ConsumerForkCurlTest.php +++ b/test/ConsumerForkCurlTest.php @@ -4,6 +4,8 @@ class ConsumerForkCurlTest extends ConsumerTransportTestCase { + use HttpRetryTransportTests; + protected function consumer(): string { return 'fork_curl'; diff --git a/test/ConsumerSocketTest.php b/test/ConsumerSocketTest.php index 3fc2903..f21b708 100644 --- a/test/ConsumerSocketTest.php +++ b/test/ConsumerSocketTest.php @@ -7,45 +7,57 @@ class ConsumerSocketTest extends ConsumerTransportTestCase { + use HttpRetryTransportTests; + protected function consumer(): string { return 'socket'; } - public static function retryableStatuses(): array + public static function incompleteResponseCases(): array { - return ['server failure' => [503], 'rate limit' => [429]]; + return [ + 'missing header terminator' => ["HTTP/1.1 200 OK\r\nContent-Length: 0\r\n", 0.1], + 'oversized unterminated header' => ["HTTP/1.1 200 OK\r\nX-Padding: " . str_repeat('a', 70 * 1024), 6.0], + ]; } - #[DataProvider('retryableStatuses')] - public function testRetrySendsTheCompleteBatch(int $status): void + #[DataProvider('incompleteResponseCases')] + public function testIncompleteResponseKeepsBatchForRetry(string $response, float $timeout): void { $this->server->stop(); - $this->server = new LocalHttpServer([['status' => $status], ['status' => 200]]); - $client = $this->client(['maximum_backoff_duration' => 200]); - $previousTimeout = ini_set('default_socket_timeout', '1'); + $this->server = new LocalHttpServer([['raw_response' => $response]]); + $client = $this->client(['timeout' => $timeout]); + $previousTimeout = ini_set('default_socket_timeout', '6'); try { - self::assertTrue($client->capture(['distinctId' => 'some-user', 'event' => 'retried event'])); - self::assertTrue($client->flush()); - $requests = $this->server->requests(); - self::assertCount(2, $requests); - self::assertSame('POST /batch/ HTTP/1.1', $requests[1]['requestLine']); - self::assertSame($requests[0]['body'], $requests[1]['body']); - $payload = json_decode($requests[1]['body'], true, 512, JSON_THROW_ON_ERROR); - self::assertSame('retried event', $payload['batch'][0]['event']); + self::assertTrue($client->capture(['distinctId' => 'some-user', 'event' => 'incomplete response'])); + self::assertFalse($client->flush()); + self::assertCount(1, $this->server->requests()); self::assertTrue($client->flush()); - self::assertCount(2, $this->server->requests()); + [$failed, $resent] = $this->server->requests(); + self::assertSame($failed['body'], $resent['body']); } finally { ini_set('default_socket_timeout', $previousTimeout); } } - public function testShortTimeout(): void + public static function defaultReadTimeouts(): array + { + return ['bounded default' => ['60'], 'unlimited default' => ['-1']]; + } + + #[DataProvider('defaultReadTimeouts')] + public function testShortTimeout(string $defaultTimeout): void { - $client = $this->client(['timeout' => 0.01]); - self::assertTrue($client->capture(['distinctId' => 'some-user', 'event' => 'short timeout'])); - self::assertSame('short timeout', $this->flushEvent($client)['event']); + $previousTimeout = ini_set('default_socket_timeout', $defaultTimeout); + try { + $client = $this->client(['timeout' => 0.01]); + self::assertTrue($client->capture(['distinctId' => 'some-user', 'event' => 'short timeout'])); + self::assertSame('short timeout', $this->flushEvent($client)['event']); + } finally { + ini_set('default_socket_timeout', $previousTimeout); + } } public function testBatchSizeOneConnectionErrorReturnsFalse(): void diff --git a/test/HttpClientTest.php b/test/HttpClientTest.php index 43c37b1..d239d5b 100644 --- a/test/HttpClientTest.php +++ b/test/HttpClientTest.php @@ -5,6 +5,8 @@ use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\TestCase; use PostHog\HttpClient; +use Symfony\Component\Clock\Clock; +use Symfony\Component\Clock\MockClock; class RetryAfterHttpClient extends HttpClient { @@ -63,6 +65,7 @@ public static function retryAfterCases(): array 'malformed' => [['Retry-After: not-a-date'], null], 'past date' => [['Retry-After: Sat, 01 Jan 2000 00:00:00 GMT'], 0], 'case insensitive' => [['retry-after: 2'], 2000], + 'beyond integer range' => [['Retry-After: 999999999999999999999999'], intdiv(PHP_INT_MAX, 1000) * 1000], ]; } @@ -119,6 +122,36 @@ public function testHttpResponsesAndRetryPolicy(int $status, bool $retry, int $a } } + public static function retryDelayCases(): array + { + return [ + 'backoff without retry after' => [[], 100], + 'retry after zero' => [['Retry-After' => '0'], 0], + 'retry after below ceiling' => [['Retry-After' => '2'], 2000], + 'retry after above ceiling' => [['Retry-After' => '3600'], 10000], + 'retry after beyond integer range' => [['Retry-After' => '999999999999999999999999'], 10000], + ]; + } + + #[DataProvider('retryDelayCases')] + public function testRetryDelayHonorsRetryAfterUpToCeiling(array $headers, int $delayMilliseconds): void + { + $server = new LocalHttpServer([['status' => 503, 'headers' => $headers], ['status' => 200]]); + $clock = new MockClock(); + $started = $clock->now(); + $previousClock = Clock::get(); + Clock::set($clock); + try { + $client = new HttpClient($server->address(), useSsl: false, maximumBackoffDuration: 201); + $response = $client->sendRequest('/batch/', '{"batch":[]}', [], ['timeout' => 5000]); + $this->assertSame(200, $response->getResponseCode()); + $this->assertEquals($started->modify('+' . $delayMilliseconds . ' milliseconds'), $clock->now()); + } finally { + Clock::set($previousClock); + $server->stop(); + } + } + public function testConditionalGetPreservesEtagAndSeparatesResponseHeaders(): void { $server = new LocalHttpServer([ diff --git a/test/HttpRetryTransportTests.php b/test/HttpRetryTransportTests.php new file mode 100644 index 0000000..b5ec4ec --- /dev/null +++ b/test/HttpRetryTransportTests.php @@ -0,0 +1,76 @@ + [408, [], true, 2, 100], + 'server error' => [500, [], true, 2, 100], + 'bad gateway' => [502, [], true, 2, 100], + 'unavailable' => [503, [], true, 2, 100], + 'gateway timeout' => [504, [], true, 2, 100], + 'rate limit' => [429, [], true, 2, 100], + 'retry after seconds' => [429, ['rEtRy-AfTeR' => '2'], true, 2, 2000], + 'retry after zero' => [503, ['Retry-After' => '0'], true, 2, 0], + 'invalid retry after' => [503, ['Retry-After' => 'not-a-date'], true, 2, 100], + 'past retry after' => [503, ['Retry-After' => 'Sat, 01 Jan 2000 00:00:00 GMT'], true, 2, 0], + 'retry after beyond integer range' => [429, ['Retry-After' => '999999999999999999999999'], true, 2, 10000], + 'long response headers' => [ + 429, ['X-Padding' => str_repeat('a', 3000), 'Retry-After' => '2'], true, 2, 2000, + ], + 'gzip retry' => [503, [], true, 2, 100, true], + 'bounded retries' => [503, [], false, 2, 100], + 'bad request' => [400, [], false, 1, 0], + 'unauthorized' => [401, [], false, 1, 0], + 'forbidden' => [403, [], false, 1, 0], + 'too large' => [413, [], false, 1, 0], + ]; + } + + #[DataProvider('httpRetryCases')] + public function testHttpDeliveryRetryPolicy( + int $status, + array $headers, + bool $success, + int $attempts, + int $delayMilliseconds, + bool $compressed = false + ): void { + $this->server->stop(); + $responses = array_fill(0, $attempts - 1, ['status' => $status, 'headers' => $headers]); + $responses[] = ['status' => $success ? 200 : $status, 'headers' => $headers]; + $this->server = new LocalHttpServer($responses); + $client = $this->client(['maximum_backoff_duration' => 200, 'compress_request' => $compressed]); + $clock = new MockClock(); + $started = $clock->now(); + $previousClock = Clock::get(); + Clock::set($clock); + + try { + self::assertTrue($client->capture(['distinctId' => 'some-user', 'event' => 'retried event'])); + self::assertSame($success, $client->flush()); + $requests = $this->server->requests(); + self::assertCount($attempts, $requests); + foreach ($requests as $request) { + self::assertSame('POST /batch/ HTTP/1.1', $request['requestLine']); + self::assertSame($requests[0]['body'], $request['body']); + self::assertSame($compressed ? 'gzip' : null, $request['headers']['content-encoding'] ?? null); + } + self::assertEquals($started->modify('+' . $delayMilliseconds . ' milliseconds'), $clock->now()); + $retained = !$success && $attempts > 1; + self::assertTrue($client->flush()); + $requests = $this->server->requests(); + self::assertCount($attempts + (int) $retained, $requests); + self::assertSame($requests[0]['body'], end($requests)['body']); + } finally { + Clock::set($previousClock); + } + } +} diff --git a/test/QueueConsumerTest.php b/test/QueueConsumerTest.php index 0f4fe01..dd4dcf0 100644 --- a/test/QueueConsumerTest.php +++ b/test/QueueConsumerTest.php @@ -8,6 +8,8 @@ use PostHog\Consumer\LibCurl; use PostHog\Consumer\Socket; use PostHog\QueueConsumer; +use Symfony\Component\Clock\Clock; +use Symfony\Component\Clock\MockClock; class QueueConsumerTest extends TestCase { @@ -180,22 +182,46 @@ public function testSocketConnectionFailureDropsBatch(): void $this->assertSame([], $this->queuedItems($consumer)); } - public function testRetainedFailedBatchIsRetriedBeforeNewerEvents(): void + public static function retainedBatchRetryCases(): array + { + return [ + 'automatic flush waits for cooldown' => [0, false, false], + 'automatic flush resumes after cooldown' => [10, false, true], + 'explicit flush ignores cooldown' => [0, true, true], + ]; + } + + #[DataProvider('retainedBatchRetryCases')] + public function testRetainedBatchRetry(int $secondsLater, bool $explicitFlush, bool $retried): void { $first = $this->message('first'); $second = $this->message('second'); $third = $this->message('third'); $consumer = new QueueConsumerTestConsumer( - [QueueConsumerTestConsumer::retryableFailure(), true], + [QueueConsumerTestConsumer::retryableFailure(), true, true], ['batch_size' => 2] ); + $clock = new MockClock(); + $previousClock = Clock::get(); + Clock::set($clock); + + try { + $this->assertTrue($consumer->enqueue($first)); + $this->assertFalse($consumer->enqueue($second)); + $clock->sleep($secondsLater); + $this->assertTrue($consumer->enqueue($third)); + if ($explicitFlush) { + $this->assertTrue($consumer->flush()); + } + } finally { + Clock::set($previousClock); + } - $this->assertTrue($consumer->enqueue($first)); - $this->assertFalse($consumer->enqueue($second)); - $this->assertTrue($consumer->enqueue($third)); - - $this->assertSame([[$first, $second], [$first, $second], [$third]], $consumer->flushedBatches); - $this->assertSame([], $consumer->queuedItems()); + $this->assertSame( + $retried ? [[$first, $second], [$first, $second], [$third]] : [[$first, $second]], + $consumer->flushedBatches + ); + $this->assertSame($retried ? [] : [$first, $second, $third], $consumer->queuedItems()); } #[DataProvider('libCurlFailureQueueBehaviorCases')] @@ -224,7 +250,9 @@ public static function libCurlFailureQueueBehaviorCases(): array { return [ 'network failure keeps batch queued' => ['network-failure', false, 0, 28, true], - 'http failure drops batch' => ['http-failure', '{"status":0}', 500, 0, false], + 'server error keeps batch queued' => ['server-error', '{"status":0}', 500, 0, true], + 'rate limit keeps batch queued' => ['rate-limit', '{"status":0}', 429, 0, true], + 'bad request drops batch' => ['bad-request', '{"status":0}', 400, 0, false], 'payload too large drops batch' => ['payload-too-large', '{"status":0}', 413, 0, false], ]; } diff --git a/test/fixtures/http-server.php b/test/fixtures/http-server.php index 2575c00..68b9f62 100644 --- a/test/fixtures/http-server.php +++ b/test/fixtures/http-server.php @@ -44,7 +44,7 @@ foreach ($response['headers'] ?? [] as $name => $value) { $headers .= "$name: $value\r\n"; } - $wire = "HTTP/1.1 $status Response\r\n" . $headers + $wire = $response['raw_response'] ?? "HTTP/1.1 $status Response\r\n" . $headers . 'Content-Length: ' . strlen($body) . "\r\nConnection: close\r\n\r\n" . $body; while ($wire !== '') { $written = fwrite($connection, $wire); @@ -53,5 +53,13 @@ } $wire = substr($wire, $written); } + if (isset($response['raw_response'])) { + while (!feof($connection)) { + $chunk = fread($connection, 2048); + if ($chunk === false || $chunk === '') { + break; + } + } + } fclose($connection); }