diff --git a/.changeset/olive-vans-throw.md b/.changeset/olive-vans-throw.md new file mode 100644 index 0000000..43db926 --- /dev/null +++ b/.changeset/olive-vans-throw.md @@ -0,0 +1,5 @@ +--- +"posthog-php": patch +--- + +Fix socket retries failing after reconnecting or crashing when reconnection fails. diff --git a/lib/Consumer/Socket.php b/lib/Consumer/Socket.php index 52cecb7..71d1c97 100644 --- a/lib/Consumer/Socket.php +++ b/lib/Consumer/Socket.php @@ -113,14 +113,15 @@ private function createSocket() */ private function makeRequest($socket, $req, $retry = true) { - $bytes_written = 0; $bytes_total = strlen($req); - $closed = false; // Retries with exponential backoff until success $backoff = 100; // Set initial waiting time to 100ms while (true) { + $bytes_written = 0; + $closed = false; + // Send request to server while (!$closed && $bytes_written < $bytes_total) { try { @@ -173,6 +174,9 @@ private function makeRequest($socket, $req, $retry = true) // Retry uploading... $backoff *= 2; $socket = $this->createSocket(); + if (!$socket) { + return false; + } } return false; diff --git a/test/ConsumerSocketTest.php b/test/ConsumerSocketTest.php index 38479ce..141ad7c 100644 --- a/test/ConsumerSocketTest.php +++ b/test/ConsumerSocketTest.php @@ -2,6 +2,7 @@ namespace PostHog\Test; +use PHPUnit\Framework\Attributes\DataProvider; use RuntimeException; class ConsumerSocketTest extends ConsumerTransportTestCase @@ -60,6 +61,64 @@ public function testHttpFailureDropsBatchAndLogsWhenDebugging(): void self::assertSame(['[PostHog][Socket] invalid payload'], $errorMessages); } + public static function retryCases(): array + { + return [ + 'server error' => [503, false], + 'rate limited gzip' => [429, true], + ]; + } + + #[DataProvider('retryCases')] + public function testRetryResendsCompleteRequest(int $status, bool $compressed): void + { + $this->server->stop(); + $this->server = new LocalHttpServer([['status' => $status], ['status' => 200]]); + $client = $this->client([ + 'maximum_backoff_duration' => 200, + 'compress_request' => $compressed, + ]); + self::assertTrue($client->capture([ + 'distinctId' => 'some-user', + 'event' => 'retried event', + 'properties' => ['text' => str_repeat('retry payload ', 100)], + ])); + + $previousTimeout = ini_set('default_socket_timeout', '1'); + try { + self::assertTrue($client->flush()); + } finally { + ini_set('default_socket_timeout', $previousTimeout); + } + + $requests = $this->server->requests(); + self::assertCount(2, $requests); + self::assertSame($requests[0], $requests[1]); + $payload = json_decode($requests[1]['body'], true, 512, JSON_THROW_ON_ERROR); + self::assertSame('retried event', $payload['batch'][0]['event']); + self::assertSame( + $compressed ? 'gzip' : null, + $requests[1]['headers']['content-encoding'] ?? null + ); + self::assertTrue($client->flush()); + self::assertCount(2, $this->server->requests()); + } + + public function testFailedReconnectReturnsFalseWithoutInvalidStreamAccess(): void + { + $this->server->stop(); + $this->server = new LocalHttpServer([['status' => 503, 'close_server' => true]]); + $client = $this->client([ + 'maximum_backoff_duration' => 200, + 'timeout' => 0.05, + 'debug' => false, + ]); + self::assertTrue($client->capture(['distinctId' => 'some-user', 'event' => 'failed retry'])); + self::assertFalse($client->flush()); + self::assertCount(1, $this->server->requests()); + self::assertTrue($client->flush()); + } + public function testConnectionError(): void { $this->server->stop(); diff --git a/test/fixtures/http-server.php b/test/fixtures/http-server.php index 2575c00..5774ad9 100644 --- a/test/fixtures/http-server.php +++ b/test/fixtures/http-server.php @@ -38,6 +38,10 @@ ], JSON_THROW_ON_ERROR) . "\n", FILE_APPEND); $response = array_shift($responses) ?? []; + $closeServer = $response['close_server'] ?? false; + if ($closeServer) { + fclose($server); + } $status = $response['status'] ?? 200; $body = $response['body'] ?? '{"status":1}'; $headers = ''; @@ -54,4 +58,7 @@ $wire = substr($wire, $written); } fclose($connection); + if ($closeServer) { + break; + } }