From eb174e7f49cb2d0f93d18eb2a40596c90f417a74 Mon Sep 17 00:00:00 2001 From: Deeka Wong <8337659+huangdijia@users.noreply.github.com> Date: Thu, 20 Aug 2026 23:27:25 +0800 Subject: [PATCH] fix(sentry): make transport push non-blocking and resilient to consumer failure --- src/sentry/publish/sentry.php | 5 +- src/sentry/src/Transport/CoHttpTransport.php | 164 +++++++++++---- tests/Sentry/CoHttpTransportTest.php | 200 +++++++++++++++++++ 3 files changed, 334 insertions(+), 35 deletions(-) create mode 100644 tests/Sentry/CoHttpTransportTest.php diff --git a/src/sentry/publish/sentry.php b/src/sentry/publish/sentry.php index fb99e720a..bb51e9106 100644 --- a/src/sentry/publish/sentry.php +++ b/src/sentry/publish/sentry.php @@ -184,7 +184,10 @@ // Transport configuration 'transport_channel_size' => (int) env('SENTRY_TRANSPORT_CHANNEL_SIZE', 512), 'transport_concurrent_limit' => (int) env('SENTRY_TRANSPORT_CONCURRENT_LIMIT', 100), - 'transport_timeout' => (float) env('SENTRY_TRANSPORT_TIMEOUT', 0), + // The max seconds to wait when pushing an event into the transport channel + // `<= 0` means non-blocking (skip the event immediately when the channel is full), + // `> 0` means wait at most N seconds for the channel to have capacity. + 'transport_timeout' => (float) env('SENTRY_TRANSPORT_TIMEOUT', 1.0), 'http_timeout' => (float) env('SENTRY_HTTP_TIMEOUT', 2.0), ]; diff --git a/src/sentry/src/Transport/CoHttpTransport.php b/src/sentry/src/Transport/CoHttpTransport.php index ffa5e658e..7d205ea0c 100644 --- a/src/sentry/src/Transport/CoHttpTransport.php +++ b/src/sentry/src/Transport/CoHttpTransport.php @@ -12,6 +12,7 @@ namespace FriendsOfHyperf\Sentry\Transport; use Hyperf\Contract\ConfigInterface; +use Hyperf\Contract\StdoutLoggerInterface; use Hyperf\Coordinator\Constants; use Hyperf\Coordinator\CoordinatorManager; use Hyperf\Coroutine\Concurrent; @@ -42,7 +43,7 @@ class CoHttpTransport implements TransportInterface protected int $channelSize = 65535; - protected float $timeout = -1; + protected float $timeout = 0; public function __construct( protected ContainerInterface $container, @@ -58,28 +59,102 @@ public function __construct( $this->concurrent = new Concurrent($concurrentLimit); } - $timeout = (float) $config->get('sentry.transport_timeout', -1); - if ($timeout > 0) { - $this->timeout = $timeout; - } + $this->timeout = $this->resolvePushTimeout(); } public function send(Event $event): Result { $this->loop(); - $chan = $this->chan; - // push event to channel, if timeout is set, it will wait for the specified time - $result = $chan?->push($event, $this->timeout) ? ResultStatus::success() : ResultStatus::skipped(); + if ($this->chan === null) { + $this->logWarning('Sentry transport channel is not available, the event will be skipped.'); + + return new Result(ResultStatus::skipped(), $event); + } + + if (! $this->pushEvent($event)) { + $this->logWarning('Sentry transport channel is full, the event will be skipped.'); + + return new Result(ResultStatus::skipped(), $event); + } - return new Result($result, $event); + return new Result(ResultStatus::success(), $event); } public function close(?int $timeout = null): Result { + $timeout ??= 1; + + $chan = $this->chan; + + if ($chan === null) { + return new Result(ResultStatus::success()); + } + + $startedAt = microtime(true); + + while (! $chan->isEmpty()) { + if ((microtime(true) - $startedAt) >= $timeout) { + break; + } + + msleep(100); + } + + $this->closeChannel(); + return new Result(ResultStatus::success()); } + /** + * Resolve the timeout used when pushing an event into the channel from + * the `sentry.transport_timeout` config. + * + * Swoole Channel::push() semantics: -1 (and any other non-positive + * value) blocks until space is available, while a positive value waits + * at most N seconds. We map a non-positive config to 0 so that + * pushEvent() takes the non-blocking path and senders can never be + * suspended indefinitely. + */ + protected function resolvePushTimeout(): float + { + $timeout = (float) $this->container->get(ConfigInterface::class)->get('sentry.transport_timeout', 0); + + // A non-positive timeout means non-blocking: when the channel is full, + // the push returns false immediately and the event is skipped. + return $timeout <= 0 ? 0.0 : $timeout; + } + + /** + * Push an event into the channel. + * + * When the configured push timeout is not positive the push is + * non-blocking: if the channel is full the event is skipped without + * suspending the caller coroutine. + */ + protected function pushEvent(Event $event): bool + { + $chan = $this->chan; + + if ($chan === null) { + return false; + } + + if ($this->timeout <= 0) { + if ($chan->isFull()) { + return false; + } + + // Swoole Channel::push() treats a non-positive timeout as "block + // until space is available", so guard against a full channel and + // bound the push with a tiny timeout to cover the race where + // another producer fills the channel right after the check. + return $chan->push($event, 0.001); + } + + return $chan->push($event, $this->timeout); + } + protected function loop(): void { if ($this->workerExited) { @@ -93,38 +168,44 @@ protected function loop(): void $this->chan = new Channel($this->channelSize); Coroutine::create(function () { - while (true) { - $transport = $this->makeHttpTransport(); - $logger = $this->clientBuilder?->getLogger(); - + try { while (true) { - /** @var null|Event|false $event */ - $event = $this->chan?->pop(); + $transport = $this->makeHttpTransport(); + $logger = $this->clientBuilder?->getLogger(); - if (! $event) { - break 2; - } + while (true) { + /** @var null|Event|false $event */ + $event = $this->chan?->pop(); - try { - $callable = static fn () => $transport->send($event); - if ($this->concurrent !== null) { - $this->concurrent->create($callable); - } else { - Coroutine::create($callable); + if (! $event) { + break 2; + } + + try { + $callable = static fn () => $transport->send($event); + if ($this->concurrent !== null) { + $this->concurrent->create($callable); + } else { + Coroutine::create($callable); + } + } catch (Throwable $e) { + $logger?->error('Failed to send event to Sentry: ' . $e->getMessage(), ['exception' => $e]); + $transport->close(); + + break; + } finally { + // Prevent memory leak + $event = null; } - } catch (Throwable $e) { - $logger?->error('Failed to send event to Sentry: ' . $e->getMessage(), ['exception' => $e]); - $transport->close(); - - break; - } finally { - // Prevent memory leak - $event = null; } } + } catch (Throwable $e) { + // The consumer died (e.g. makeHttpTransport() failed), close the + // channel so that send() can rebuild it on the next call. + $this->clientBuilder?->getLogger()?->error('Failed to initialize Sentry transport: ' . $e->getMessage(), ['exception' => $e]); + } finally { + $this->closeChannel(); } - - $this->closeChannel(); }); $this->workerWatcher ??= Coroutine::create(function () { @@ -164,4 +245,19 @@ protected function closeChannel(): void $this->chan = null; } } + + protected function logWarning(string $message): void + { + $logger = $this->clientBuilder?->getLogger(); + + if ($logger === null) { + try { + $logger = $this->container->get(StdoutLoggerInterface::class); + } catch (Throwable) { + // Ignore, the logger is not available. + } + } + + $logger?->warning($message); + } } diff --git a/tests/Sentry/CoHttpTransportTest.php b/tests/Sentry/CoHttpTransportTest.php new file mode 100644 index 000000000..418998c31 --- /dev/null +++ b/tests/Sentry/CoHttpTransportTest.php @@ -0,0 +1,200 @@ +resolvePushTimeout(); + } + + public function getTimeoutForTest(): float + { + return $this->timeout; + } + + public function pushEventForTest(Event $event): bool + { + return $this->pushEvent($event); + } + + public function getChanForTest(): ?Channel + { + return $this->chan; + } + + public function setChanForTest(?Channel $chan): void + { + $this->chan = $chan; + } + + protected function loop(): void + { + // no-op: avoid spawning the consumer and worker watcher coroutines in tests + } +} + +function createCoHttpTransportTestable(array $overrides = []): CoHttpTransportTestable +{ + $config = new Config([ + 'sentry' => array_merge([ + 'transport_channel_size' => 512, + 'transport_concurrent_limit' => 100, + 'transport_timeout' => 0, + ], $overrides), + ]); + + $container = Mockery::mock(ContainerInterface::class); + $container->shouldReceive('get')->with(ConfigInterface::class)->andReturn($config); + + return new CoHttpTransportTestable($container); +} + +function runTestInCoroutine(callable $callback): void +{ + $exception = null; + + \Swoole\Coroutine\run(function () use ($callback, &$exception) { + try { + $callback(); + } catch (Throwable $e) { + $exception = $e; + } + }); + + if ($exception !== null) { + throw $exception; + } +} + +test('resolvePushTimeout returns non-blocking timeout for non-positive values', function () { + runTestInCoroutine(function () { + $transport = createCoHttpTransportTestable(['transport_timeout' => 0]); + expect($transport->resolvePushTimeoutForTest())->toBe(0.0); + + $transport = createCoHttpTransportTestable(['transport_timeout' => -1]); + expect($transport->resolvePushTimeoutForTest())->toBe(0.0); + + $transport = createCoHttpTransportTestable(['transport_timeout' => 1.5]); + expect($transport->resolvePushTimeoutForTest())->toBe(1.5); + }); +}); + +test('constructor maps non-positive transport_timeout to a non-blocking push timeout', function () { + runTestInCoroutine(function () { + $transport = createCoHttpTransportTestable(['transport_timeout' => -1]); + expect($transport->getTimeoutForTest())->toBe(0.0); + + $transport = createCoHttpTransportTestable(['transport_timeout' => 2]); + expect($transport->getTimeoutForTest())->toBe(2.0); + }); +}); + +test('send returns skipped without blocking when the channel is full', function () { + runTestInCoroutine(function () { + $transport = createCoHttpTransportTestable([ + 'transport_channel_size' => 1, + 'transport_timeout' => 0, + ]); + $transport->setChanForTest(new Channel(1)); + + // Fill the channel first, the next push must not block and must fail. + expect($transport->pushEventForTest(Event::createEvent()))->toBeTrue(); + + $result = $transport->send(Event::createEvent()); + expect($result->getStatus())->toBe(ResultStatus::skipped()); + }); +}); + +test('send returns skipped when the channel is missing', function () { + runTestInCoroutine(function () { + $transport = createCoHttpTransportTestable(); + $transport->setChanForTest(null); + + $result = $transport->send(Event::createEvent()); + expect($result->getStatus())->toBe(ResultStatus::skipped()); + }); +}); + +test('send returns success when the channel has capacity', function () { + runTestInCoroutine(function () { + $transport = createCoHttpTransportTestable([ + 'transport_channel_size' => 2, + 'transport_timeout' => 0, + ]); + $transport->setChanForTest(new Channel(2)); + + $result = $transport->send(Event::createEvent()); + expect($result->getStatus())->toBe(ResultStatus::success()); + }); +}); + +test('close waits for the channel to drain and then closes it', function () { + runTestInCoroutine(function () { + $transport = createCoHttpTransportTestable([ + 'transport_channel_size' => 8, + 'transport_timeout' => 0, + ]); + $chan = new Channel(8); + $transport->setChanForTest($chan); + + $chan->push(Event::createEvent()); + $chan->push(Event::createEvent()); + + // Drain the channel slowly in a background coroutine. + Coroutine::create(function () use ($chan) { + while ($chan->pop(1) !== false) { + msleep(50); + } + }); + + $startedAt = microtime(true); + $result = $transport->close(1); + $elapsed = microtime(true) - $startedAt; + + expect($result->getStatus())->toBe(ResultStatus::success()); + // It waited for the backlog to be drained instead of closing immediately. + expect($elapsed)->toBeGreaterThanOrEqual(0.04); + // It did not block beyond the given timeout. + expect($elapsed)->toBeLessThan(2.0); + expect($transport->getChanForTest())->toBeNull(); + }); +}); + +test('close returns success when the channel does not exist', function () { + runTestInCoroutine(function () { + $transport = createCoHttpTransportTestable(); + $transport->setChanForTest(null); + + $result = $transport->close(1); + expect($result->getStatus())->toBe(ResultStatus::success()); + }); +});