From 991f0de1c0638c4391c84e753ca73172efadd682 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Wed, 21 Oct 2026 18:43:41 +0200 Subject: [PATCH 1/2] fix: honor ingestion transport timeouts --- .changeset/wet-rings-notice.md | 5 + lib/Consumer/ForkCurl.php | 2 + lib/Consumer/LibCurl.php | 3 +- test/ConsumerCurlTimeoutTest.php | 172 ++++++++++++++++++++++++ test/fixtures/curl-timeout-receiver.php | 38 ++++++ 5 files changed, 219 insertions(+), 1 deletion(-) create mode 100644 .changeset/wet-rings-notice.md create mode 100644 test/ConsumerCurlTimeoutTest.php create mode 100644 test/fixtures/curl-timeout-receiver.php diff --git a/.changeset/wet-rings-notice.md b/.changeset/wet-rings-notice.md new file mode 100644 index 00000000..bd3856dd --- /dev/null +++ b/.changeset/wet-rings-notice.md @@ -0,0 +1,5 @@ +--- +"posthog-php": patch +--- + +Honor the configured ingestion timeout in the libcurl and forked curl transports. diff --git a/lib/Consumer/ForkCurl.php b/lib/Consumer/ForkCurl.php index 9d4f8e3d..ef72380d 100644 --- a/lib/Consumer/ForkCurl.php +++ b/lib/Consumer/ForkCurl.php @@ -64,7 +64,9 @@ public function flushBatch($messages) $path = "/batch/"; $url = $protocol . $this->host . $path; + $timeout = escapeshellarg((string) (($this->options['timeout'] ?? 10000) / 1000)); $cmd = "curl -X POST -H 'Content-Type: application/json'"; + $cmd .= " --connect-timeout " . $timeout . " --max-time " . $timeout; $tmpfname = ""; if ($this->compress_request) { diff --git a/lib/Consumer/LibCurl.php b/lib/Consumer/LibCurl.php index 24fd727f..e90a9295 100644 --- a/lib/Consumer/LibCurl.php +++ b/lib/Consumer/LibCurl.php @@ -33,7 +33,8 @@ public function __construct($apiKey, $options = [], ?HttpClient $httpClient = nu $this->maximum_backoff_duration, $this->compress_request, $this->debug(), - $this->options['error_handler'] ?? null + $this->options['error_handler'] ?? null, + (int) ($this->options['timeout'] ?? 10000) ); } diff --git a/test/ConsumerCurlTimeoutTest.php b/test/ConsumerCurlTimeoutTest.php new file mode 100644 index 00000000..49870351 --- /dev/null +++ b/test/ConsumerCurlTimeoutTest.php @@ -0,0 +1,172 @@ + ['LibCurl', false, true], + 'forkcurl synchronous' => ['ForkCurl', true, true], + 'forkcurl spawned' => ['ForkCurl', false, true], + ]; + } + + #[DataProvider('stalledTransports')] + public function testConfiguredTimeoutClosesStalledConnection(string $transport, bool $debug, bool $verify): void + { + $this->runReceiver($transport, ['timeout' => 50, 'debug' => $debug, + 'verify_batch_events_request' => $verify], false); + } + + public function testLibCurlWithoutVerificationDoesNotWaitEvenWithZeroTimeout(): void + { + $this->runReceiver('LibCurl', ['timeout' => 0, 'debug' => false, + 'verify_batch_events_request' => false], false); + } + + public static function zeroTimeoutTransports(): array + { + return ['libcurl' => ['LibCurl'], 'forkcurl' => ['ForkCurl']]; + } + + #[DataProvider('zeroTimeoutTransports')] + public function testZeroTimeoutAllowsDelayedResponse(string $transport): void + { + $this->runReceiver($transport, ['timeout' => 0, 'debug' => true], true); + } + + public function testLibCurlDefaultTimeoutIsTenSeconds(): void + { + $consumer = new LibCurl('test-key'); + $client = (new ReflectionProperty(LibCurl::class, 'httpClient'))->getValue($consumer); + self::assertSame(10000, (new ReflectionProperty($client, 'curlTimeoutMilliseconds'))->getValue($client)); + } + + public static function forkCurlTimeoutArguments(): array + { + return [ + 'default' => [[], '10'], + 'milliseconds' => [['timeout' => 50], '0.05'], + 'fractional seconds' => [['timeout' => 1250], '1.25'], + 'unlimited' => [['timeout' => 0], '0'], + ]; + } + + #[DataProvider('forkCurlTimeoutArguments')] + public function testForkCurlSetsConnectAndTotalTimeoutArguments(array $options, string $expected): void + { + $directory = sys_get_temp_dir() . '/posthog-curl-timeout-' . bin2hex(random_bytes(8)); + mkdir($directory); + $arguments = $directory . '/arguments'; + file_put_contents($directory . '/curl', "#!/bin/sh\nprintf '%s\\n' \"\$@\" > " . escapeshellarg($arguments)); + chmod($directory . '/curl', 0700); + $process = null; + try { + $process = $this->startConsumer('ForkCurl', array_merge($options, ['debug' => true]), $pipes, $directory); + self::assertTrue($this->waitForExit($process), 'ForkCurl command exceeded the watchdog'); + self::assertSame('true', trim(stream_get_contents($pipes[1]))); + $args = file($arguments, FILE_IGNORE_NEW_LINES); + foreach (['--connect-timeout', '--max-time'] as $argument) { + $index = array_search($argument, $args, true); + self::assertNotFalse($index, 'Missing ' . $argument); + self::assertSame($expected, $args[$index + 1]); + } + } finally { + $this->stopProcess($process, $pipes ?? []); + foreach (glob($directory . '/*') as $file) { + unlink($file); + } + rmdir($directory); + } + } + + private function runReceiver(string $transport, array $options, bool $respond): void + { + $server = proc_open( + [PHP_BINARY, __DIR__ . '/fixtures/curl-timeout-receiver.php', $respond ? 'respond' : 'stall'], + [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['file', '/dev/null', 'a']], + $serverPipes + ); + self::assertIsResource($server); + $process = null; + try { + stream_set_timeout($serverPipes[1], 5); + $address = trim(fgets($serverPipes[1])); + self::assertNotSame('', $address, 'Timeout receiver did not become ready'); + $options['host'] = $address; + $process = $this->startConsumer($transport, $options, $pipes); + if (($options['verify_batch_events_request'] ?? true) === false) { + self::assertTrue($this->waitForExit($process), 'Unverified request exceeded the watchdog'); + self::assertSame('"retryable_failure"', trim(stream_get_contents($pipes[1]))); + return; + } + self::assertSame("received\n", fgets($serverPipes[1]), 'Receiver must observe the request'); + stream_set_blocking($serverPipes[1], false); + $output = ''; + $deadline = microtime(true) + 3; + do { + $output .= stream_get_contents($serverPipes[1]); + if (strpos($output, "closed\n") !== false) { + break; + } + usleep(10000); + } while (microtime(true) < $deadline); + self::assertStringContainsString("closed\n", $output, 'Connection exceeded the three-second watchdog'); + self::assertTrue($this->waitForExit($process), 'Consumer exceeded the three-second watchdog'); + $result = trim(stream_get_contents($pipes[1])); + $expected = $respond || ($transport === 'ForkCurl' && !$options['debug']) + ? 'true' : '"retryable_failure"'; + self::assertSame($expected, $result); + } finally { + $this->stopProcess($server, $serverPipes); + $this->stopProcess($process, $pipes ?? []); + } + } + + private function startConsumer(string $transport, array $options, ?array &$pipes, ?string $path = null) + { + $options = array_merge(['ssl' => false, 'maximum_backoff_duration' => 100], $options); + $script = 'require ' . var_export(dirname(__DIR__) . '/vendor/autoload.php', true) . ';'; + if ($path !== null) { + $script .= 'putenv(' . var_export('PATH=' . $path . ':' . getenv('PATH'), true) . ');'; + } + $script .= '$consumer = new \\PostHog\\Consumer\\' . $transport . '("test-key", ' + . var_export($options, true) . ');' + . 'echo json_encode($consumer->flushBatch([["event" => "timeout", "distinct_id" => "user"]]));'; + return proc_open( + [PHP_BINARY, '-d', 'display_errors=stderr', '-r', $script], + [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['file', '/dev/null', 'a']], + $pipes + ); + } + + private function waitForExit($process): bool + { + $deadline = microtime(true) + 3; + do { + if (!proc_get_status($process)['running']) { + return true; + } + usleep(10000); + } while (microtime(true) < $deadline); + return false; + } + + private function stopProcess($process, array $pipes): void + { + if (is_resource($process)) { + proc_terminate($process); + foreach ($pipes as $pipe) { + fclose($pipe); + } + proc_close($process); + } + } +} diff --git a/test/fixtures/curl-timeout-receiver.php b/test/fixtures/curl-timeout-receiver.php new file mode 100644 index 00000000..f471d443 --- /dev/null +++ b/test/fixtures/curl-timeout-receiver.php @@ -0,0 +1,38 @@ + 0 && ($chunk = fread($connection, $length)) !== false && $chunk !== '') { + $length -= strlen($chunk); +} +echo "received\n"; +flush(); +if (($argv[1] ?? '') === 'respond') { + usleep(200000); + fwrite($connection, "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}"); +} else { + while (!feof($connection)) { + if (fread($connection, 8192) === false) { + break; + } + } +} +fclose($connection); +fclose($server); +echo "closed\n"; From 441365204771748071eb618792c68271c5e6748e Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Wed, 21 Oct 2026 19:23:11 +0200 Subject: [PATCH 2/2] fix: keep fractional ingestion timeouts bounded --- .changeset/wet-rings-notice.md | 2 +- lib/Consumer/ForkCurl.php | 2 +- lib/Consumer/LibCurl.php | 2 +- test/ConsumerCurlTimeoutTest.php | 40 +++++++++++++++++++++++++++----- 4 files changed, 37 insertions(+), 9 deletions(-) diff --git a/.changeset/wet-rings-notice.md b/.changeset/wet-rings-notice.md index bd3856dd..e90d1a61 100644 --- a/.changeset/wet-rings-notice.md +++ b/.changeset/wet-rings-notice.md @@ -2,4 +2,4 @@ "posthog-php": patch --- -Honor the configured ingestion timeout in the libcurl and forked curl transports. +Honor configured ingestion timeouts in the libcurl and forked curl transports, rounding positive fractional milliseconds up so they remain bounded. diff --git a/lib/Consumer/ForkCurl.php b/lib/Consumer/ForkCurl.php index ef72380d..a529399b 100644 --- a/lib/Consumer/ForkCurl.php +++ b/lib/Consumer/ForkCurl.php @@ -64,7 +64,7 @@ public function flushBatch($messages) $path = "/batch/"; $url = $protocol . $this->host . $path; - $timeout = escapeshellarg((string) (($this->options['timeout'] ?? 10000) / 1000)); + $timeout = escapeshellarg((string) (ceil($this->options['timeout'] ?? 10000) / 1000)); $cmd = "curl -X POST -H 'Content-Type: application/json'"; $cmd .= " --connect-timeout " . $timeout . " --max-time " . $timeout; diff --git a/lib/Consumer/LibCurl.php b/lib/Consumer/LibCurl.php index e90a9295..04d98c0f 100644 --- a/lib/Consumer/LibCurl.php +++ b/lib/Consumer/LibCurl.php @@ -34,7 +34,7 @@ public function __construct($apiKey, $options = [], ?HttpClient $httpClient = nu $this->compress_request, $this->debug(), $this->options['error_handler'] ?? null, - (int) ($this->options['timeout'] ?? 10000) + (int) ceil($this->options['timeout'] ?? 10000) ); } diff --git a/test/ConsumerCurlTimeoutTest.php b/test/ConsumerCurlTimeoutTest.php index 49870351..b6b9f143 100644 --- a/test/ConsumerCurlTimeoutTest.php +++ b/test/ConsumerCurlTimeoutTest.php @@ -25,6 +25,20 @@ public function testConfiguredTimeoutClosesStalledConnection(string $transport, 'verify_batch_events_request' => $verify], false); } + public static function positiveFractionalTransports(): array + { + return [ + 'libcurl' => ['LibCurl', false], + 'forkcurl' => ['ForkCurl', true], + ]; + } + + #[DataProvider('positiveFractionalTransports')] + public function testPositiveFractionalTimeoutDoesNotWaitIndefinitely(string $transport, bool $debug): void + { + $this->runReceiver($transport, ['timeout' => 0.5, 'debug' => $debug], false, expectRequest: false); + } + public function testLibCurlWithoutVerificationDoesNotWaitEvenWithZeroTimeout(): void { $this->runReceiver('LibCurl', ['timeout' => 0, 'debug' => false, @@ -42,11 +56,23 @@ public function testZeroTimeoutAllowsDelayedResponse(string $transport): void $this->runReceiver($transport, ['timeout' => 0, 'debug' => true], true); } - public function testLibCurlDefaultTimeoutIsTenSeconds(): void + public static function libCurlTimeoutValues(): array + { + return [ + 'default' => [[], 10000], + 'unlimited' => [['timeout' => 0], 0], + 'sub-millisecond' => [['timeout' => 0.5], 1], + 'fractional milliseconds' => [['timeout' => 1.5], 2], + 'whole milliseconds' => [['timeout' => 50], 50], + ]; + } + + #[DataProvider('libCurlTimeoutValues')] + public function testLibCurlTimeoutUsesWholeMilliseconds(array $options, int $expected): void { - $consumer = new LibCurl('test-key'); + $consumer = new LibCurl('test-key', $options); $client = (new ReflectionProperty(LibCurl::class, 'httpClient'))->getValue($consumer); - self::assertSame(10000, (new ReflectionProperty($client, 'curlTimeoutMilliseconds'))->getValue($client)); + self::assertSame($expected, (new ReflectionProperty($client, 'curlTimeoutMilliseconds'))->getValue($client)); } public static function forkCurlTimeoutArguments(): array @@ -54,6 +80,8 @@ public static function forkCurlTimeoutArguments(): array return [ 'default' => [[], '10'], 'milliseconds' => [['timeout' => 50], '0.05'], + 'sub-millisecond' => [['timeout' => 0.5], '0.001'], + 'fractional milliseconds' => [['timeout' => 1.5], '0.002'], 'fractional seconds' => [['timeout' => 1250], '1.25'], 'unlimited' => [['timeout' => 0], '0'], ]; @@ -87,7 +115,7 @@ public function testForkCurlSetsConnectAndTotalTimeoutArguments(array $options, } } - private function runReceiver(string $transport, array $options, bool $respond): void + private function runReceiver(string $transport, array $options, bool $respond, bool $expectRequest = true): void { $server = proc_open( [PHP_BINARY, __DIR__ . '/fixtures/curl-timeout-receiver.php', $respond ? 'respond' : 'stall'], @@ -102,8 +130,8 @@ private function runReceiver(string $transport, array $options, bool $respond): self::assertNotSame('', $address, 'Timeout receiver did not become ready'); $options['host'] = $address; $process = $this->startConsumer($transport, $options, $pipes); - if (($options['verify_batch_events_request'] ?? true) === false) { - self::assertTrue($this->waitForExit($process), 'Unverified request exceeded the watchdog'); + if (!$expectRequest || ($options['verify_batch_events_request'] ?? true) === false) { + self::assertTrue($this->waitForExit($process), 'Consumer exceeded the watchdog'); self::assertSame('"retryable_failure"', trim(stream_get_contents($pipes[1]))); return; }