From 69cac0e58028d6cec706ce5e8d458f34f49a93fd Mon Sep 17 00:00:00 2001 From: Pavel Vondrak Date: Fri, 21 Aug 2026 13:50:56 +0200 Subject: [PATCH 1/2] feat!: upgrade to bunny 0.6 Bunny 0.6 is a rewrite: it dropped the sync/async duality and runs on a ReactPHP event loop using PHP Fibers, so every Client/Channel call has to execute inside a Fiber, and failures are reported by throwing instead of by returning *OkFrame protocol objects. All broker I/O is funnelled through the new Connection::run(Closure), which wraps the operation in React\Async\await(async(...)), and does so even when a Fiber is already current - running the operation on the calling Fiber instead leaves PHP unable to switch out of contexts that forbid it, such as a signal handler. Callers keep a plain synchronous API and must never drive the loop themselves. The *OkFrame return checks in SetupAction and ConsumerRunner become try/catch blocks that pass the cause through as $previous, and the instanceof PromiseInterface guards are gone now that there is a single return type. ConsumerRunner no longer polls Client::run() in a while loop; it awaits a Deferred settled by the message-count limit or a Loop timer. Bunny reports asynchronous failures as 'error'/'close' events, which Evenement silently drops when nobody listens, so the runner listens on both and wraps Consumer::consume() in a try/catch - without that, a throwing consumer, a broker-closed channel or a lost connection left the consume loop blocked forever with nothing but an unhandled promise rejection on stderr. Both events are needed: a broker that closes the channel emits 'close' and then 'error' with the reply code and text, while a connection that goes away emits only 'close'. Since the first rejection is the one that counts, the 'close' fallback is held back by a tick so that the error saying why gets there first. Every settle goes through Loop::futureTick(), because resuming the awaiting Fiber from inside a delivery callback's own Fiber never lets React's scheduler hand the result back to run()'s caller. While connected, Bunny keeps a heartbeat timer on the loop that would keep a php-fpm worker or console process alive after the work is done. The new DisconnectConnection subscriber closes the connection on kernel.terminate and console.terminate, covering producers in both HTTP and CLI contexts. BunnyConnection also drops a cached channel on its 'close' event, since 0.6 throws ChannelException('Channel is closed') for every later call on a channel the broker closed and a single 404 publish would otherwise poison the connection for the rest of the process, and it replaces a client whose connect() failed, because Bunny leaves such a client reporting itself as connected while refusing to connect again. One upstream limitation remains: bunny 0.6.0-alpha.4 cannot recover in-process from a first connection attempt that failed, even with a brand-new Client. A connection that was established and then lost reconnects on the next operation. BREAKING CHANGE: * Requires bunny/bunny 0.6, currently available only as an alpha pre-release, so consuming projects have to allow that stability. * The read_write_timeout option is gone, as Bunny 0.6 has no read/write timeout. The DSN query parameter is merely ignored, but the YAML key now fails the container build and has to be removed. * RabbitMQ\Connection gained run(Closure): mixed - templated over what the closure returns, so callers keep the type they passed in - and getChannel() and getTransactionalChannel() now return Bunny\ChannelInterface rather than Bunny\Channel, so custom implementations need updating. Signed-off-by: Pavel Vondrak --- composer.json | 9 +- docs/Installation.md | 26 +++ docs/Producing.md | 14 +- docs/Setup.md | 2 +- phpstan-baseline.neon | 8 +- phpunit.xml.dist | 2 +- src/Configuration/Connection.php | 20 -- src/ConsumerRunner.php | 211 ++++++++++++------ src/DependencyInjection/Configuration.php | 6 - src/EventListener/DisconnectConnection.php | 71 ++++++ src/Exception/CannotCreateChannel.php | 6 - src/Exception/ConfigurationFailed.php | 26 ++- src/Exception/ConnectionFailed.php | 5 + src/Exception/OperationFailed.php | 6 - src/RabbitMQ/Binding.php | 6 +- src/RabbitMQ/BunnyConnection.php | 122 ++++++---- src/RabbitMQ/Connection.php | 25 ++- src/RabbitMQ/Exchange.php | 6 +- src/RabbitMQ/Message.php | 6 +- .../Operation/AcknowledgeOperation.php | 18 +- src/RabbitMQ/Operation/GetOperation.php | 26 +-- src/RabbitMQ/Operation/PublishOperation.php | 88 ++++---- src/RabbitMQ/Operation/RejectOperation.php | 20 +- src/RabbitMQ/Queue.php | 6 +- src/Resources/config/services.yaml | 2 + src/SetupAction.php | 102 +++++---- tests/ConsumerRunnerFailureTest.php | 107 +++++++++ tests/ConsumerRunnerTest.php | 81 +++++-- .../DisconnectConnectionTest.php | 95 ++++++++ tests/RabbitMQ/BunnyConnectionTest.php | 92 ++++++++ tests/RabbitMQ/ConfigurationTest.php | 1 - tests/RabbitMQ/ConfigurationTest.yaml | 2 +- tests/RabbitMQ/ThrowingConsumer.php | 48 ++++ 33 files changed, 943 insertions(+), 322 deletions(-) create mode 100644 src/EventListener/DisconnectConnection.php create mode 100644 tests/ConsumerRunnerFailureTest.php create mode 100644 tests/EventListener/DisconnectConnectionTest.php create mode 100644 tests/RabbitMQ/BunnyConnectionTest.php create mode 100644 tests/RabbitMQ/ThrowingConsumer.php diff --git a/composer.json b/composer.json index c62ba5f..af4970c 100644 --- a/composer.json +++ b/composer.json @@ -5,12 +5,15 @@ "type": "library", "require": { "php": "^8.3", - "bunny/bunny": "^0.5.0", + "bunny/bunny": "^0.6.0-alpha.4", "myclabs/php-enum": "^1.8.4", - "react/promise": "^2.0", + "react/async": "^4.2", + "react/event-loop": "^1.2", + "react/promise": "^3", "symfony/config": "^7.4", "symfony/console": "^7.4", "symfony/dependency-injection": "^7.4", + "symfony/event-dispatcher": "^7.4", "symfony/framework-bundle": "^7.4", "symfony/http-kernel": "^7.4", "symfony/yaml": "^7.4" @@ -26,6 +29,8 @@ "phpunit/phpunit": "^12.5", "shipmonk/composer-dependency-analyser": "^1.4" }, + "minimum-stability": "alpha", + "prefer-stable": true, "autoload": { "psr-4": { "Cdn77\\RabbitMQBundle\\": "src/" diff --git a/docs/Installation.md b/docs/Installation.md index 8c3461c..1d1c386 100644 --- a/docs/Installation.md +++ b/docs/Installation.md @@ -6,6 +6,32 @@ Add as a dependency via Composer: composer require cdn77/rabbitmq-bundle ``` +> **Note:** This bundle requires `bunny/bunny` `0.6`, which is currently only available as a +> pre-release (`alpha`). Until a stable `0.6.0` is tagged, your project must allow that stability, +> e.g. by setting in your `composer.json`: +> +> ```json +> { +> "minimum-stability": "alpha", +> "prefer-stable": true +> } +> ``` +> +> Bunny `0.6` runs on a [ReactPHP](https://reactphp.org/) event loop using PHP Fibers, so the +> bundle pulls in `react/event-loop`, `react/async` and `react/promise` as well. All broker I/O is +> driven synchronously for you — see [Producing](Producing.md) for the one caveat in long-lived CLI +> scripts. + +> **Upgrading from a Bunny `0.5` release of this bundle:** Bunny `0.6` has no read/write timeout, so +> the `read_write_timeout` option is gone. A `read_write_timeout` query parameter in your DSN is +> simply ignored, but the YAML key now fails the container build with +> `Unrecognized option "read_write_timeout" under "rabbitmq"` — drop it from your configuration. +> +> Note also that Bunny `0.6` cannot recover within the same process from a *first* connection +> attempt that failed (the broker being unreachable at start-up); every later attempt on that +> connection reports `ConnectionFailed`. Losing an already established connection is handled and +> reconnects on the next operation. + If you're not using Symfony Flex, you will also need to enable the bundle by adding `Cdn77RabbitMQBundle` to `bundles.php`, that is required by `registerBundles()` in your `Kernel`: ```php diff --git a/docs/Producing.md b/docs/Producing.md index 69ccf24..e1144b9 100644 --- a/docs/Producing.md +++ b/docs/Producing.md @@ -9,11 +9,15 @@ use Cdn77\RabbitMQBundle\RabbitMQ\Operation\PublishOperation; final class ExampleProducer { + /** @var Connection */ + private $connection; + /** @var PublishOperation */ private $publishOperation; public function __construct(Connection $connection, PublishOperation $publishOperation) { + $this->connection = $connection; $this->publishOperation = $publishOperation; } @@ -28,7 +32,7 @@ final class ExampleProducer $message->makeTransient(); $this->publishOperation->handle( - $this->connection + $this->connection, $message, $routingKey, 'default_exchange', // Exchange to send the message to @@ -36,3 +40,11 @@ final class ExampleProducer } } ``` + +> **Heads up (Bunny 0.6):** while connected, Bunny keeps a heartbeat timer on the ReactPHP event +> loop, which would otherwise keep the php-fpm worker or console process alive after the work is +> done. The bundle ships a `DisconnectConnection` subscriber that closes the connection on +> `kernel.terminate` (after the HTTP response is flushed) and on `console.terminate` (after a +> command returns), so HTTP and console producers are handled automatically. If you publish from a +> context that dispatches neither event (e.g. a bare script using the event loop directly), call +> `$connection->disconnect()` yourself when you're done. diff --git a/docs/Setup.md b/docs/Setup.md index 33bdd62..b163008 100644 --- a/docs/Setup.md +++ b/docs/Setup.md @@ -4,7 +4,7 @@ This bundle uses `rabbitmq` container extension key and is able to merge configu These configuration options are available to setup connection to your RabbitMQ instance. -Example DSN: `amqp://username:password@host:1234/vhost?heartbeat=60&connection_timeout=10&read_write_timeout=3` +Example DSN: `amqp://username:password@host:1234/vhost?heartbeat=60&connection_timeout=10` ```yaml rabbitmq: diff --git a/phpstan-baseline.neon b/phpstan-baseline.neon index 1a2a25b..e975ce0 100644 --- a/phpstan-baseline.neon +++ b/phpstan-baseline.neon @@ -3,7 +3,7 @@ parameters: - message: '#^Cannot cast mixed to int\.$#' identifier: cast.int - count: 3 + count: 2 path: src/Configuration/Connection.php - @@ -55,7 +55,7 @@ parameters: path: src/RabbitMQ/Binding.php - - message: '#^Parameter \#3 \$arguments of class Cdn77\\RabbitMQBundle\\RabbitMQ\\Binding constructor expects array\, mixed given\.$#' + message: '#^Parameter \#3 \$arguments of class Cdn77\\RabbitMQBundle\\RabbitMQ\\Binding constructor expects array\, mixed given\.$#' identifier: argument.type count: 1 path: src/RabbitMQ/Binding.php @@ -85,7 +85,7 @@ parameters: path: src/RabbitMQ/Exchange.php - - message: '#^Parameter \#6 \$arguments of class Cdn77\\RabbitMQBundle\\RabbitMQ\\Exchange constructor expects array\, mixed given\.$#' + message: '#^Parameter \#6 \$arguments of class Cdn77\\RabbitMQBundle\\RabbitMQ\\Exchange constructor expects array\, mixed given\.$#' identifier: argument.type count: 1 path: src/RabbitMQ/Exchange.php @@ -109,7 +109,7 @@ parameters: path: src/RabbitMQ/Queue.php - - message: '#^Parameter \#5 \$arguments of class Cdn77\\RabbitMQBundle\\RabbitMQ\\Queue constructor expects array\, mixed given\.$#' + message: '#^Parameter \#5 \$arguments of class Cdn77\\RabbitMQBundle\\RabbitMQ\\Queue constructor expects array\, mixed given\.$#' identifier: argument.type count: 1 path: src/RabbitMQ/Queue.php diff --git a/phpunit.xml.dist b/phpunit.xml.dist index 8bb32cc..18983b5 100644 --- a/phpunit.xml.dist +++ b/phpunit.xml.dist @@ -8,7 +8,7 @@ bootstrap="tests/bootstrap.php" > - + tests diff --git a/src/Configuration/Connection.php b/src/Configuration/Connection.php index 8b371d4..8ab1c96 100644 --- a/src/Configuration/Connection.php +++ b/src/Configuration/Connection.php @@ -10,7 +10,6 @@ final class Connection { private const int DEFAULT_HEARTBEAT = 60; private const int DEFAULT_CONNECTION_TIMEOUT = 3; - private const int DEFAULT_READ_WRITE_TIMEOUT = 5; /** @var string */ private $host; @@ -33,9 +32,6 @@ final class Connection /** @var int */ private $connectionTimeout; - /** @var int */ - private $readWriteTimeout; - public function __construct( string $host, int $port, @@ -44,7 +40,6 @@ public function __construct( string|null $password, int $heartbeat = self::DEFAULT_HEARTBEAT, int $connectionTimeout = self::DEFAULT_CONNECTION_TIMEOUT, - int $readWriteTimeout = self::DEFAULT_READ_WRITE_TIMEOUT, ) { $this->host = $host; $this->port = $port; @@ -53,7 +48,6 @@ public function __construct( $this->password = $password; $this->heartbeat = $heartbeat; $this->connectionTimeout = $connectionTimeout; - $this->readWriteTimeout = $readWriteTimeout; } /** @param mixed[] $configuration */ @@ -76,13 +70,6 @@ public static function fromDI(array $configuration): self $new->connectionTimeout = (int) $configuration[Configuration::KEY_CONFIGURATION_CONNECTION_TIMEOUT]; } - if ( - isset($configuration[Configuration::KEY_CONFIGURATION_READ_WRITE_TIMEOUT]) - && ! isset($dsn->getParameters()[Configuration::KEY_CONFIGURATION_READ_WRITE_TIMEOUT]) - ) { - $new->readWriteTimeout = (int) $configuration[Configuration::KEY_CONFIGURATION_READ_WRITE_TIMEOUT]; - } - return $new; } @@ -99,8 +86,6 @@ public static function fromDsn(Dsn $dsn): self (int) ($parameters[Configuration::KEY_CONFIGURATION_HEARTBEAT] ?? self::DEFAULT_HEARTBEAT), (int) ($parameters[Configuration::KEY_CONFIGURATION_CONNECTION_TIMEOUT] ?? self::DEFAULT_CONNECTION_TIMEOUT), - (int) ($parameters[Configuration::KEY_CONFIGURATION_READ_WRITE_TIMEOUT] - ?? self::DEFAULT_READ_WRITE_TIMEOUT), ); } @@ -138,9 +123,4 @@ public function getConnectionTimeout(): int { return $this->connectionTimeout; } - - public function getReadWriteTimeout(): int - { - return $this->readWriteTimeout; - } } diff --git a/src/ConsumerRunner.php b/src/ConsumerRunner.php index fa65313..132524a 100644 --- a/src/ConsumerRunner.php +++ b/src/ConsumerRunner.php @@ -4,16 +4,17 @@ namespace Cdn77\RabbitMQBundle; -use Bunny\Channel; -use Bunny\Client; use Bunny\Message; -use Bunny\Protocol\MethodBasicQosOkFrame; use Cdn77\RabbitMQBundle\Exception\ConfigurationFailed; +use Cdn77\RabbitMQBundle\Exception\ConnectionFailed; use Cdn77\RabbitMQBundle\RabbitMQ\Connection; -use Cdn77\RabbitMQBundle\RabbitMQ\Consumer\Configuration; use Cdn77\RabbitMQBundle\RabbitMQ\Consumer\Consumer; +use React\EventLoop\Loop; +use React\Promise\Deferred; +use Throwable; -use function microtime; +use function React\Async\async; +use function React\Async\await; final class ConsumerRunner { @@ -30,73 +31,149 @@ public function __construct(Connection $connection) public function run(Consumer $consumer): void { - $channel = $this->connection->getChannel(); - $consumerConfiguration = $consumer->getConfiguration(); + $configuration = $consumer->getConfiguration(); + $this->processedMessageCount = 0; - $qosOk = $channel->qos($consumerConfiguration->getPrefetchSize(), $consumerConfiguration->getPrefetchCount()); - if (! $qosOk instanceof MethodBasicQosOkFrame) { - throw ConfigurationFailed::invalidPrefetchValues(); + if (! $this->hasAnyMessageLeft($configuration->getMaxMessages(), $this->processedMessageCount)) { + return; } - $this->createConsumerOnChannel($channel, $consumer); - $this->consumerRun($channel, $consumer); - } - - private function createConsumerOnChannel(Channel $channel, Consumer $consumer): void - { - $channel->consume( - function (Message $message, Channel $channel, Client $client) use ($consumer): void { - $consumerConfig = $consumer->getConfiguration(); - $consumer->consume($message); - - $this->processedMessageCount++; - - if ($this->hasAnyMessageLeft($consumerConfig->getMaxMessages(), $this->processedMessageCount)) { - return; + $this->connection->run(function () use ($consumer, $configuration): void { + $channel = $this->connection->getChannel(); + + try { + $channel->qos($configuration->getPrefetchSize(), $configuration->getPrefetchCount()); + } catch (Throwable $error) { + throw ConfigurationFailed::invalidPrefetchValues($error); + } + + /** @var Deferred $stopped */ + $stopped = new Deferred(); + $stopping = false; + + // Settle on a future tick, never straight from a delivery callback: doing it inline + // would resume the Fiber awaiting below from inside the callback's own Fiber, leaving + // React's scheduler with no way to hand the result back to run()'s caller. The flag has + // to be set right away though: Bunny hands over the next buffered delivery as soon as + // the callback returns, long before a future tick gets to run. + $stop = static function () use (&$stopping, $stopped): void { + $stopping = true; + + Loop::futureTick(static fn () => $stopped->resolve(null)); + }; + $fail = static function (Throwable $error) use (&$stopping, $stopped): void { + $stopping = true; + + Loop::futureTick(static fn () => $stopped->reject($error)); + }; + + $consumeOk = $channel->consume( + async(function (Message $message) use ( + $consumer, + $configuration, + $channel, + $stop, + $fail, + &$stopping, + ): void { + try { + // Further messages may already be in flight (prefetch) once the run is + // over - be it the limit being reached or the consumer having failed. + // Reject the ones we still get handed rather than only skipping them, so + // they go back to the queue right away instead of sitting unacknowledged - + // and invisible to other consumers - until the connection goes away. + // Whatever Bunny has buffered but not yet delivered is dropped by + // basic.cancel below and only the broker can put those back, which it does + // once this channel or connection closes. + if ( + $stopping + || ! $this->hasAnyMessageLeft( + $configuration->getMaxMessages(), + $this->processedMessageCount, + ) + ) { + $channel->nack($message, false, true); + + return; + } + + $consumer->consume($message); + + $this->processedMessageCount++; + + if ( + $this->hasAnyMessageLeft( + $configuration->getMaxMessages(), + $this->processedMessageCount, + ) + ) { + return; + } + + $stop(); + } catch (Throwable $error) { + // The callback runs in its own Fiber, so throwing here would only surface + // as an unhandled promise rejection. Hand the failure to the awaited + // promise instead to let it propagate out of run(). + $fail($error); + } + }), + $configuration->getQueueName(), + ); + + // Bunny reports channel and connection level failures as 'error'/'close' events rather + // than by throwing, and an unlistened event is silently dropped. Without these the + // await() below would keep blocking after e.g. the broker closed the channel or the + // connection was lost. Registered only now that consume() is through - it reports its + // own failures by throwing, and a rejection here would have nothing awaiting it yet. + // A broker-closed channel emits both events: 'close' first, then 'error' carrying a + // ChannelException with the reply code and text. Only the first rejection counts, so + // hold this fallback back by a tick and let the one that says why go first. Nothing + // else settles the promise when the channel goes without an error - a lost connection + // closes it silently - so the fallback still gets there. + $closed = static fn () => Loop::futureTick( + static fn () => $fail(ConnectionFailed::channelClosed()), + ); + $channel->on('error', $fail); + $channel->once('close', $closed); + + $maxSeconds = $configuration->getMaxSeconds(); + $timer = $maxSeconds !== null ? Loop::addTimer($maxSeconds, $stop) : null; + $stoppedCleanly = false; + + try { + await($stopped->promise()); + + $stoppedCleanly = true; + } finally { + if ($timer !== null) { + Loop::cancelTimer($timer); } - $client->stop(); - }, - $consumer->getConfiguration()->getQueueName(), - '', - false, - false, - false, - false, - [], - ); - } - - private function consumerRun(Channel $channel, Consumer $consumer): void - { - $consumerConfiguration = $consumer->getConfiguration(); - $startTime = microtime(true); - - while ($this->shouldContinue($startTime, $consumerConfiguration)) { - $channel->getClient()->run($consumerConfiguration->getMaxSeconds()); - } - } - - private function shouldContinue(float $startTime, Configuration $consumerConfiguration): bool - { - return $this->isInfinitelyRepeated($consumerConfiguration) || - ! $this->isLimitReached($startTime, $consumerConfiguration); - } - - private function isInfinitelyRepeated(Configuration $consumerConfiguration): bool - { - return $consumerConfiguration->getMaxSeconds() === null && $consumerConfiguration->getMaxMessages() === null; - } - - private function isLimitReached(float $startTime, Configuration $consumerConfiguration): bool - { - return ! $this->hasAnyTimeLeft($consumerConfiguration->getMaxSeconds(), $startTime) - || ! $this->hasAnyMessageLeft($consumerConfiguration->getMaxMessages(), $this->processedMessageCount); - } - - private function hasAnyTimeLeft(float|null $maxSeconds, float $startTime): bool - { - return $maxSeconds === null || microtime(true) < $startTime + $maxSeconds; + // The channel is cached and reused, so leave neither this run's listeners nor its + // consumer behind on it - the latter would keep delivering into a callback whose + // promise is already settled, silently swallowing those messages. + $channel->removeListener('error', $fail); + $channel->removeListener('close', $closed); + + try { + // No-wait, so that losing the connection right here cannot leave this blocked + // for good: Bunny does not reject a pending protocol wait when the socket goes + // away, and the promise that watched for channel failures has settled already. + // Nothing depends on the broker's confirmation anyway - Bunny stops delivering + // for this consumer tag the moment cancel() returns, and messages the broker + // still sends until it processes the frame are requeued when the channel or the + // connection closes. + $channel->cancel($consumeOk->consumerTag, true); + } catch (Throwable $error) { + // A channel the broker already closed can be neither used nor cancelled, and + // saying so must not bury the failure that ended the run. + if ($stoppedCleanly) { + throw $error; + } + } + } + }); } private function hasAnyMessageLeft(int|null $maxMessages, int $processedMessageCount): bool diff --git a/src/DependencyInjection/Configuration.php b/src/DependencyInjection/Configuration.php index b89fdd2..c46cdff 100644 --- a/src/DependencyInjection/Configuration.php +++ b/src/DependencyInjection/Configuration.php @@ -16,7 +16,6 @@ final class Configuration implements ConfigurationInterface public const string KEY_CONFIGURATION_DSN = 'dsn'; public const string KEY_CONFIGURATION_HEARTBEAT = 'heartbeat'; public const string KEY_CONFIGURATION_CONNECTION_TIMEOUT = 'connection_timeout'; - public const string KEY_CONFIGURATION_READ_WRITE_TIMEOUT = 'read_write_timeout'; public const string KEY_CONFIGURATION_EXCHANGES = 'exchanges'; public const string KEY_CONFIGURATION_QUEUES = 'queues'; public const string KEY_EXCHANGE_NAME = 'name'; @@ -35,7 +34,6 @@ final class Configuration implements ConfigurationInterface private const string DEFAULT_DSN = 'amqp://127.0.0.1/'; private const int DEFAULT_HEARTBEAT = 60; private const int DEFAULT_TIMEOUT = 10; - private const int DEAFULT_READ_WRITE_TIMEOUT = 3; public function getConfigTreeBuilder(): TreeBuilder { @@ -63,10 +61,6 @@ private function configureConnection(ArrayNodeDefinition $rootNode): void $rootNode->children() ->scalarNode(self::KEY_CONFIGURATION_CONNECTION_TIMEOUT) ->defaultValue(self::DEFAULT_TIMEOUT); - - $rootNode->children() - ->scalarNode(self::KEY_CONFIGURATION_READ_WRITE_TIMEOUT) - ->defaultValue(self::DEAFULT_READ_WRITE_TIMEOUT); } private function configureExchanges(ArrayNodeDefinition $rootNode): void diff --git a/src/EventListener/DisconnectConnection.php b/src/EventListener/DisconnectConnection.php new file mode 100644 index 0000000..3c87595 --- /dev/null +++ b/src/EventListener/DisconnectConnection.php @@ -0,0 +1,71 @@ +connection = $connection; + } + + /** @return array */ + public static function getSubscribedEvents(): array + { + return [ + KernelEvents::TERMINATE => 'disconnect', + ConsoleEvents::TERMINATE => 'disconnect', + ]; + } + + public function disconnect(object $event): void + { + if ($this->isInterruptedBySignal($event)) { + // Closing the connection properly is out of the question here, so stop the loop + // instead. Bunny's socket and heartbeat stay registered on it - what matters is that + // both of the ways they could still keep the process alive are shut: the `run()` that + // `React\Async` resumes from its own shutdown function returns, and `Loop::$stopped`, + // which this sets, is what keeps React's shutdown autorun from starting the loop + // again. Without it the process never reaches the end of its own exit(). The broker + // requeues whatever stayed unacknowledged once the socket goes with the process. + Loop::stop(); + + return; + } + + $this->connection->disconnect(); + } + + /** + * Symfony dispatches `console.terminate` from inside its `SIGINT`/`SIGTERM` handler as well, and + * PHP forbids switching Fibers in a signal handler - broker I/O from there aborts with a + * FiberError no matter which Fiber, if any, the signal happened to land on. So ask the event: + * being on a Fiber or not says nothing about it, and a normal termination - `kernel.terminate` + * included, which never carries a signal - has to disconnect either way. + */ + private function isInterruptedBySignal(object $event): bool + { + return $event instanceof ConsoleTerminateEvent && $event->getInterruptingSignal() !== null; + } +} diff --git a/src/Exception/CannotCreateChannel.php b/src/Exception/CannotCreateChannel.php index b723294..6c7ef0c 100644 --- a/src/Exception/CannotCreateChannel.php +++ b/src/Exception/CannotCreateChannel.php @@ -6,12 +6,6 @@ use RuntimeException; -use function sprintf; - final class CannotCreateChannel extends RuntimeException implements Exception { - public static function gotInvalidType(string $expected, string $actual): self - { - return new self(sprintf('Expected "%s", got "%s"', $expected, $actual)); - } } diff --git a/src/Exception/ConfigurationFailed.php b/src/Exception/ConfigurationFailed.php index c0b50aa..480f53d 100644 --- a/src/Exception/ConfigurationFailed.php +++ b/src/Exception/ConfigurationFailed.php @@ -8,28 +8,32 @@ use Cdn77\RabbitMQBundle\RabbitMQ\Exchange; use Cdn77\RabbitMQBundle\RabbitMQ\Queue; use RuntimeException; +use Throwable; use function sprintf; final class ConfigurationFailed extends RuntimeException implements Exception { - public static function invalidPrefetchValues(): self + public static function invalidPrefetchValues(Throwable|null $previous = null): self { - return new self('Could not set prefetch-size/prefetch-count'); + return new self('Could not set prefetch-size/prefetch-count', 0, $previous); } - public static function cannotDeclareExchange(Exchange $exchange): self + public static function cannotDeclareExchange(Exchange $exchange, Throwable|null $previous = null): self { - return new self(sprintf('Could not declare exchange %s', $exchange->getName())); + return new self(sprintf('Could not declare exchange %s', $exchange->getName()), 0, $previous); } - public static function cannotDeclareQueue(Queue $queue): self + public static function cannotDeclareQueue(Queue $queue, Throwable|null $previous = null): self { - return new self(sprintf('Could not declare queue %s', $queue->getName())); + return new self(sprintf('Could not declare queue %s', $queue->getName()), 0, $previous); } - public static function cannotBindExchange(Exchange $exchange, Binding $binding): self - { + public static function cannotBindExchange( + Exchange $exchange, + Binding $binding, + Throwable|null $previous = null, + ): self { return new self( sprintf( 'Could not bind exchange "%s" to "%s" with routing key "%s"', @@ -37,10 +41,12 @@ public static function cannotBindExchange(Exchange $exchange, Binding $binding): $binding->getBindable()->getName(), $binding->getRoutingKey(), ), + 0, + $previous, ); } - public static function cannotBindQueue(Queue $queue, Binding $binding): self + public static function cannotBindQueue(Queue $queue, Binding $binding, Throwable|null $previous = null): self { return new self( sprintf( @@ -49,6 +55,8 @@ public static function cannotBindQueue(Queue $queue, Binding $binding): self $binding->getBindable()->getName(), $binding->getRoutingKey(), ), + 0, + $previous, ); } } diff --git a/src/Exception/ConnectionFailed.php b/src/Exception/ConnectionFailed.php index 7f34ee5..f41e0a0 100644 --- a/src/Exception/ConnectionFailed.php +++ b/src/Exception/ConnectionFailed.php @@ -13,4 +13,9 @@ public static function causedBy(Throwable $previous): self { return new self('Connection to RabbitMQ failed', 0, $previous); } + + public static function channelClosed(): self + { + return new self('Channel was closed by the broker'); + } } diff --git a/src/Exception/OperationFailed.php b/src/Exception/OperationFailed.php index 989afcc..9feeddd 100644 --- a/src/Exception/OperationFailed.php +++ b/src/Exception/OperationFailed.php @@ -6,12 +6,6 @@ use RuntimeException; -use function sprintf; - final class OperationFailed extends RuntimeException implements Exception { - public static function gotInvalidType(string $expected, string $actual): self - { - return new self(sprintf('Expected "%s", got "%s"', $expected, $actual)); - } } diff --git a/src/RabbitMQ/Binding.php b/src/RabbitMQ/Binding.php index 29c7f76..ee2c8b4 100644 --- a/src/RabbitMQ/Binding.php +++ b/src/RabbitMQ/Binding.php @@ -14,10 +14,10 @@ final class Binding /** @var string */ private $routingKey; - /** @var mixed[] */ + /** @var array */ private $arguments; - /** @param mixed[] $arguments */ + /** @param array $arguments */ public function __construct(Bindable $bindable, string $routingKey, array $arguments = []) { $this->bindable = $bindable; @@ -45,7 +45,7 @@ public function getRoutingKey(): string return $this->routingKey; } - /** @return mixed[] */ + /** @return array */ public function getArguments(): array { return $this->arguments; diff --git a/src/RabbitMQ/BunnyConnection.php b/src/RabbitMQ/BunnyConnection.php index a9560fd..fa1451d 100644 --- a/src/RabbitMQ/BunnyConnection.php +++ b/src/RabbitMQ/BunnyConnection.php @@ -4,66 +4,79 @@ namespace Cdn77\RabbitMQBundle\RabbitMQ; -use Bunny\Channel; +use Bunny\ChannelInterface; use Bunny\Client; +use Bunny\Configuration as BunnyConfiguration; +use Bunny\Defaults; use Cdn77\RabbitMQBundle\Configuration; use Cdn77\RabbitMQBundle\Exception\CannotCreateChannel; use Cdn77\RabbitMQBundle\Exception\ConnectionFailed; -use React\Promise\PromiseInterface; +use Closure; use Throwable; +use function React\Async\async; +use function React\Async\await; + final class BunnyConnection implements Connection { + /** @var BunnyConfiguration */ + private $configuration; + /** @var Client */ private $client; - /** @var Channel|null */ + /** @var ChannelInterface|null */ private $channel; - /** @var Channel */ + /** @var ChannelInterface|null */ private $transactionalChannel; public function __construct(Configuration\Connection $configuration) { - $options = [ - 'host' => $configuration->getHost(), - 'port' => $configuration->getPort(), - 'vhost' => $configuration->getVhost(), - 'heartbeat' => $configuration->getHeartbeat(), - 'timeout' => $configuration->getConnectionTimeout(), - 'read_write_timeout' => $configuration->getReadWriteTimeout(), - ]; - - if ($configuration->getUser() !== null) { - $options['user'] = $configuration->getUser(); - } - - if ($configuration->getPassword() !== null) { - $options['password'] = $configuration->getPassword(); - } - - $this->client = new Client($options); + $this->configuration = new BunnyConfiguration( + host: $configuration->getHost(), + port: $configuration->getPort(), + vhost: $configuration->getVhost(), + user: $configuration->getUser() ?? Defaults::USER, + password: $configuration->getPassword() ?? Defaults::PASSWORD, + timeout: $configuration->getConnectionTimeout(), + heartbeat: (float) $configuration->getHeartbeat(), + ); + $this->client = new Client($this->configuration); } - public function getChannel(): Channel + public function getChannel(): ChannelInterface { - if ($this->channel === null) { - $this->channel = $this->createChannel(); + $channel = $this->channel; + if ($channel === null) { + $channel = $this->createChannel(); + $channel->once('close', function (): void { + $this->channel = null; + }); + + $this->channel = $channel; } - return $this->channel; + return $channel; } - public function getTransactionalChannel(): Channel + public function getTransactionalChannel(): ChannelInterface { if ($this->transactionalChannel === null) { - $this->transactionalChannel = $this->createChannel(); + $channel = $this->createChannel(); + $channel->once('close', function (): void { + $this->transactionalChannel = null; + }); try { - $this->transactionalChannel->txSelect(); + $channel->txSelect(); } catch (Throwable $exception) { throw new CannotCreateChannel('Cannot create transaction channel', 0, $exception); } + + // Cache it only once it is transactional, otherwise a retry would hand back a plain + // channel from the fast path above and fail on txCommit() instead. + $this->transactionalChannel = $channel; } return $this->transactionalChannel; @@ -71,12 +84,23 @@ public function getTransactionalChannel(): Channel public function connect(): void { - if ($this->client->isConnected()) { + if ($this->client->canDisconnect()) { return; } + // Bunny never rolls the state back when connect() fails, leaving a client that reports + // itself as connected yet refuses to connect again - so every later attempt would await a + // promise nothing can resolve. Such a client is unusable; start over with a fresh one. + if ($this->client->isConnected()) { + $this->channel = null; + $this->transactionalChannel = null; + $this->client = new Client($this->configuration); + } + try { - $this->client->connect(); + $this->run(function (): void { + $this->client->connect(); + }); } catch (Throwable $exception) { throw ConnectionFailed::causedBy($exception); } @@ -84,24 +108,40 @@ public function connect(): void public function disconnect(): void { - if (! $this->client->isConnected()) { + // Bunny only tolerates disconnect() on a fully connected client - canDisconnect() also + // rules out the connecting/disconnecting states that isConnected() reports as connected. + if (! $this->client->canDisconnect()) { return; } - $this->client->disconnect(); + $this->run(function (): void { + $this->client->disconnect(); + }); + $this->channel = null; + $this->transactionalChannel = null; } - private function createChannel(): Channel + /** + * @param Closure(): T $operation + * + * @return T + * + * @template T + */ + public function run(Closure $operation): mixed { - $this->connect(); - - $channel = $this->client->channel(); + // Always on a Fiber of its own, even when one is already current (an acknowledge issued + // from inside a consumer callback). Running the operation on the calling Fiber instead + // saves an allocation, but makes PHP unable to switch back out of contexts that forbid it + // - a signal handler above all - turning a survivable shutdown into a fatal FiberError. + return await(async($operation)()); + } - if ($channel instanceof PromiseInterface) { - throw CannotCreateChannel::gotInvalidType(Channel::class, PromiseInterface::class); - } + private function createChannel(): ChannelInterface + { + $this->connect(); - return $channel; + return $this->client->channel(); } } diff --git a/src/RabbitMQ/Connection.php b/src/RabbitMQ/Connection.php index b99798a..1ceac66 100644 --- a/src/RabbitMQ/Connection.php +++ b/src/RabbitMQ/Connection.php @@ -4,15 +4,34 @@ namespace Cdn77\RabbitMQBundle\RabbitMQ; -use Bunny\Channel; +use Bunny\ChannelInterface; +use Closure; interface Connection { - public function getChannel(): Channel; + public function getChannel(): ChannelInterface; - public function getTransactionalChannel(): Channel; + public function getTransactionalChannel(): ChannelInterface; public function connect(): void; public function disconnect(): void; + + /** + * Runs the given operation inside a Fiber so that the underlying Bunny + * client/channel calls (which suspend on the event loop) can execute. + * + * Implementations must give the operation a Fiber of its own even when one is already current, + * as it is for an acknowledge issued from inside a consumer callback: running it on the calling + * Fiber saves an allocation but leaves PHP unable to switch back out of contexts that forbid it + * - a signal handler above all - turning a survivable shutdown into a fatal FiberError. Nesting + * run() calls is therefore allowed, and each nested call gets its own Fiber. + * + * @param Closure(): T $operation + * + * @return T the value returned by $operation + * + * @template T + */ + public function run(Closure $operation): mixed; } diff --git a/src/RabbitMQ/Exchange.php b/src/RabbitMQ/Exchange.php index ec2b02a..78f3943 100644 --- a/src/RabbitMQ/Exchange.php +++ b/src/RabbitMQ/Exchange.php @@ -23,13 +23,13 @@ final class Exchange implements Bindable /** @var bool */ private $internal; - /** @var mixed[] */ + /** @var array */ private $arguments; /** @var Binding[] */ private $bindings = []; - /** @param mixed[] $arguments */ + /** @param array $arguments */ public function __construct( string $name, ExchangeType $exchangeType, @@ -95,7 +95,7 @@ public function isInternal(): bool return $this->internal; } - /** @return mixed[] */ + /** @return array */ public function getArguments(): array { return $this->arguments; diff --git a/src/RabbitMQ/Message.php b/src/RabbitMQ/Message.php index 87c6070..ac6af40 100644 --- a/src/RabbitMQ/Message.php +++ b/src/RabbitMQ/Message.php @@ -12,10 +12,10 @@ class Message /** @var string */ public $body; - /** @var mixed[] */ + /** @var array */ public $headers; - /** @param mixed[] $headers */ + /** @param array $headers */ public function __construct(string $body, array $headers = []) { $this->body = $body; @@ -27,7 +27,7 @@ public function __construct(string $body, array $headers = []) $this->headers = $headers; } - /** @param mixed[] $headers */ + /** @param array $headers */ public static function json(string $body, array $headers = []): self { return new self($body, [self::HEADER_CONTENT_TYPE => 'application/json'] + $headers); diff --git a/src/RabbitMQ/Operation/AcknowledgeOperation.php b/src/RabbitMQ/Operation/AcknowledgeOperation.php index cef1b7a..964d2db 100644 --- a/src/RabbitMQ/Operation/AcknowledgeOperation.php +++ b/src/RabbitMQ/Operation/AcknowledgeOperation.php @@ -19,12 +19,9 @@ public function __construct(Connection $connection) public function handle(Message $message): void { - $channel = $this->connection->getChannel(); - $channel->getClient()->ack( - $channel->getChannelId(), - $message->deliveryTag, - false, - ); + $this->connection->run(function () use ($message): void { + $this->connection->getChannel()->ack($message); + }); } /** @@ -33,11 +30,8 @@ public function handle(Message $message): void */ public function handleAll(Message $lastMessage): void { - $channel = $this->connection->getChannel(); - $channel->getClient()->ack( - $channel->getChannelId(), - $lastMessage->deliveryTag, - true, - ); + $this->connection->run(function () use ($lastMessage): void { + $this->connection->getChannel()->ack($lastMessage, true); + }); } } diff --git a/src/RabbitMQ/Operation/GetOperation.php b/src/RabbitMQ/Operation/GetOperation.php index 2072ac7..556c6df 100644 --- a/src/RabbitMQ/Operation/GetOperation.php +++ b/src/RabbitMQ/Operation/GetOperation.php @@ -5,9 +5,7 @@ namespace Cdn77\RabbitMQBundle\RabbitMQ\Operation; use Bunny\Message; -use Cdn77\RabbitMQBundle\Exception\OperationFailed; use Cdn77\RabbitMQBundle\RabbitMQ\Connection; -use React\Promise\PromiseInterface; final class GetOperation { @@ -22,23 +20,21 @@ public function __construct(Connection $connection) /** @return Message[] */ public function handle(string $queueName, int $maxCount): array { - $messages = []; + return $this->connection->run(function () use ($queueName, $maxCount): array { + $messages = []; + $channel = $this->connection->getChannel(); - for ($count = 0; $count < $maxCount; $count++) { - /** @var Message|PromiseInterface|null $message */ - $message = $this->connection->getChannel()->get($queueName, false); + for ($count = 0; $count < $maxCount; $count++) { + $message = $channel->get($queueName, false); - if ($message === null) { - return $messages; - } + if ($message === null) { + return $messages; + } - if ($message instanceof PromiseInterface) { - throw OperationFailed::gotInvalidType(Message::class, PromiseInterface::class); + $messages[] = $message; } - $messages[] = $message; - } - - return $messages; + return $messages; + }); } } diff --git a/src/RabbitMQ/Operation/PublishOperation.php b/src/RabbitMQ/Operation/PublishOperation.php index 3403fa0..298ceb6 100644 --- a/src/RabbitMQ/Operation/PublishOperation.php +++ b/src/RabbitMQ/Operation/PublishOperation.php @@ -14,7 +14,7 @@ final class PublishOperation private const bool MANDATORY = false; private const bool IMMEDIATE = false; - /** @param mixed[] $headers */ + /** @param array $headers */ public function handleRaw( Connection $connection, string $body, @@ -22,26 +22,30 @@ public function handleRaw( string $routingKey, string $exchange, ): void { - $connection->getChannel()->publish( - $body, - $headers, - $exchange, - $routingKey, - self::MANDATORY, - self::IMMEDIATE, - ); + $connection->run(static function () use ($connection, $body, $headers, $routingKey, $exchange): void { + $connection->getChannel()->publish( + $body, + $headers, + $exchange, + $routingKey, + self::MANDATORY, + self::IMMEDIATE, + ); + }); } public function handle(Connection $connection, Message $message, string $routingKey, string $exchange): void { - $connection->getChannel()->publish( - $message->body, - $message->headers, - $exchange, - $routingKey, - self::MANDATORY, - self::IMMEDIATE, - ); + $connection->run(static function () use ($connection, $message, $routingKey, $exchange): void { + $connection->getChannel()->publish( + $message->body, + $message->headers, + $exchange, + $routingKey, + self::MANDATORY, + self::IMMEDIATE, + ); + }); } /** @param Message[] $messages */ @@ -51,28 +55,36 @@ public function handleAll( string $routingKey, string $exchangeName, ): void { - $transactionalChannel = $connection->getTransactionalChannel(); - try { - foreach ($messages as $message) { - $transactionalChannel->publish( - $message->body, - $message->headers, - $exchangeName, - $routingKey, - self::MANDATORY, - self::IMMEDIATE, - ); - } + $connection->run(static function () use ($connection, $messages, $routingKey, $exchangeName): void { + $transactionalChannel = $connection->getTransactionalChannel(); + try { + foreach ($messages as $message) { + $transactionalChannel->publish( + $message->body, + $message->headers, + $exchangeName, + $routingKey, + self::MANDATORY, + self::IMMEDIATE, + ); + } - $transactionalChannel->txCommit(); - } catch (Throwable $exception) { - $transactionalChannel->txRollback(); + $transactionalChannel->txCommit(); + } catch (Throwable $exception) { + try { + $transactionalChannel->txRollback(); + } catch (Throwable) { + // The usual cause of the failure above is the broker closing the channel, and + // such a channel can no longer be rolled back - nor does it need to be. Keep + // reporting what actually went wrong. + } - throw new OperationFailed( - $exception->getMessage(), - $exception->getCode(), - $exception, - ); - } + throw new OperationFailed( + $exception->getMessage(), + $exception->getCode(), + $exception, + ); + } + }); } } diff --git a/src/RabbitMQ/Operation/RejectOperation.php b/src/RabbitMQ/Operation/RejectOperation.php index 0dd28dc..b43bcee 100644 --- a/src/RabbitMQ/Operation/RejectOperation.php +++ b/src/RabbitMQ/Operation/RejectOperation.php @@ -19,13 +19,9 @@ public function __construct(Connection $connection) public function handle(Message $message, bool $requeue = true): void { - $channel = $this->connection->getChannel(); - $channel->getClient()->nack( - $channel->getChannelId(), - $message->deliveryTag, - false, - $requeue, - ); + $this->connection->run(function () use ($message, $requeue): void { + $this->connection->getChannel()->nack($message, false, $requeue); + }); } /** @@ -34,12 +30,8 @@ public function handle(Message $message, bool $requeue = true): void */ public function handleAll(Message $lastMessage, bool $requeue = true): void { - $channel = $this->connection->getChannel(); - $channel->getClient()->nack( - $channel->getChannelId(), - $lastMessage->deliveryTag, - true, - $requeue, - ); + $this->connection->run(function () use ($lastMessage, $requeue): void { + $this->connection->getChannel()->nack($lastMessage, true, $requeue); + }); } } diff --git a/src/RabbitMQ/Queue.php b/src/RabbitMQ/Queue.php index 8ee54dc..1a4d6da 100644 --- a/src/RabbitMQ/Queue.php +++ b/src/RabbitMQ/Queue.php @@ -23,10 +23,10 @@ final class Queue implements Bindable /** @var bool */ private $autoDelete; - /** @var mixed[] */ + /** @var array */ private $arguments; - /** @param mixed[] $arguments */ + /** @param array $arguments */ public function __construct( string $name, bool $durable = false, @@ -84,7 +84,7 @@ public function shouldAutoDelete(): bool return $this->autoDelete; } - /** @return mixed[] */ + /** @return array */ public function getArguments(): array { return $this->arguments; diff --git a/src/Resources/config/services.yaml b/src/Resources/config/services.yaml index a156563..98e9262 100644 --- a/src/Resources/config/services.yaml +++ b/src/Resources/config/services.yaml @@ -37,3 +37,5 @@ services: Cdn77\RabbitMQBundle\ConsumerRunner: Cdn77\RabbitMQBundle\SetupAction: + + Cdn77\RabbitMQBundle\EventListener\DisconnectConnection: diff --git a/src/SetupAction.php b/src/SetupAction.php index 1fc562f..760831b 100644 --- a/src/SetupAction.php +++ b/src/SetupAction.php @@ -4,13 +4,10 @@ namespace Cdn77\RabbitMQBundle; -use Bunny\Protocol\MethodExchangeBindOkFrame; -use Bunny\Protocol\MethodExchangeDeclareOkFrame; -use Bunny\Protocol\MethodQueueBindOkFrame; -use Bunny\Protocol\MethodQueueDeclareOkFrame; use Cdn77\RabbitMQBundle\Configuration\Topology; use Cdn77\RabbitMQBundle\Exception\ConfigurationFailed; use Cdn77\RabbitMQBundle\RabbitMQ\Connection; +use Throwable; final class SetupAction { @@ -23,69 +20,78 @@ public function __construct(Connection $connection) } public function setup(Topology $topology): void + { + $this->connection->run(function () use ($topology): void { + $this->declareTopology($topology); + }); + } + + private function declareTopology(Topology $topology): void { $channel = $this->connection->getChannel(); foreach ($topology->getExchanges() as $exchange) { - $frame = $channel->exchangeDeclare( - $exchange->getName(), - $exchange->getExchangeType()->getValue(), - false, - $exchange->isDurable(), - $exchange->shouldAutoDelete(), - $exchange->isInternal(), - false, - $exchange->getArguments(), - ); - - if (! ($frame instanceof MethodExchangeDeclareOkFrame)) { - throw ConfigurationFailed::cannotDeclareExchange($exchange); + try { + $channel->exchangeDeclare( + $exchange->getName(), + $exchange->getExchangeType()->getValue(), + false, + $exchange->isDurable(), + $exchange->shouldAutoDelete(), + $exchange->isInternal(), + false, + $exchange->getArguments(), + ); + } catch (Throwable $exception) { + throw ConfigurationFailed::cannotDeclareExchange($exchange, $exception); } foreach ($exchange->getBindings() as $binding) { $boundQueue = $binding->getBindable(); - $frame = $channel->exchangeBind( - $exchange->getName(), - $boundQueue->getName(), - $binding->getRoutingKey(), - false, - $binding->getArguments(), - ); - - if (! ($frame instanceof MethodExchangeBindOkFrame)) { - throw ConfigurationFailed::cannotBindExchange($exchange, $binding); + try { + $channel->exchangeBind( + $exchange->getName(), + $boundQueue->getName(), + $binding->getRoutingKey(), + false, + $binding->getArguments(), + ); + } catch (Throwable $exception) { + throw ConfigurationFailed::cannotBindExchange($exchange, $binding, $exception); } } } foreach ($topology->getQueues() as $queue) { - $frame = $channel->queueDeclare( - $queue->getName(), - false, - $queue->isDurable(), - $queue->isExclusive(), - $queue->shouldAutoDelete(), - false, - $queue->getArguments(), - ); - - if (! ($frame instanceof MethodQueueDeclareOkFrame)) { - throw ConfigurationFailed::cannotDeclareQueue($queue); + try { + $channel->queueDeclare( + $queue->getName(), + false, + $queue->isDurable(), + $queue->isExclusive(), + $queue->shouldAutoDelete(), + false, + $queue->getArguments(), + ); + } catch (Throwable $exception) { + throw ConfigurationFailed::cannotDeclareQueue($queue, $exception); } foreach ($queue->getBindings() as $binding) { $boundQueue = $binding->getBindable(); - $frame = $channel->queueBind( - $queue->getName(), - $boundQueue->getName(), - $binding->getRoutingKey(), - false, - $binding->getArguments(), - ); - if (! ($frame instanceof MethodQueueBindOkFrame)) { - throw ConfigurationFailed::cannotBindQueue($queue, $binding); + try { + // Bunny 0.6 changed queue.bind argument order to (exchange, queue). + $channel->queueBind( + $boundQueue->getName(), + $queue->getName(), + $binding->getRoutingKey(), + false, + $binding->getArguments(), + ); + } catch (Throwable $exception) { + throw ConfigurationFailed::cannotBindQueue($queue, $binding, $exception); } } } diff --git a/tests/ConsumerRunnerFailureTest.php b/tests/ConsumerRunnerFailureTest.php new file mode 100644 index 0000000..0effa91 --- /dev/null +++ b/tests/ConsumerRunnerFailureTest.php @@ -0,0 +1,107 @@ +createPartialMock(Channel::class, ['qos']); + $channel->method('qos')->willThrowException(new ClientException('NOT_IMPLEMENTED')); + + $connection = $this->givenConnectionTo($channel); + + // A non-zero prefetch-size, the rejection RabbitMQ is guaranteed to answer with. Nothing + // is ever consumed, so the acknowledge operation is only there to build the consumer. + $consumer = new InMemoryConsumer( + new AcknowledgeOperation($connection), + new Configuration('aQueue', 1, 1), + ); + + self::expectException(ConfigurationFailed::class); + self::expectExceptionMessage('Could not set prefetch-size/prefetch-count'); + + (new ConsumerRunner($connection))->run($consumer); + } + + public function testChannelErrorOutrunsTheCloseFallback(): void + { + $channel = $this->createPartialMock(Channel::class, ['qos', 'consume', 'cancel']); + $channel->method('qos')->willReturn(new MethodBasicQosOkFrame()); + $channel->method('cancel')->willReturn(false); + $channel->method('consume')->willReturnCallback( + static function () use ($channel): MethodBasicConsumeOkFrame { + // What Bunny does with a channel.close frame, in that order and within one tick - + // and only once consume() is through, which is when the runner starts listening. + Loop::futureTick(static function () use ($channel): void { + $channel->emit('close'); + $channel->emit('error', [new ChannelException(self::CHANNEL_CLOSED, 404)]); + }); + + $consumeOk = new MethodBasicConsumeOkFrame(); + $consumeOk->consumerTag = 'aConsumerTag'; + + return $consumeOk; + }, + ); + + $connection = $this->givenConnectionTo($channel); + $consumer = new InMemoryConsumer( + new AcknowledgeOperation($connection), + new Configuration('aQueue'), + ); + + // Not ConnectionFailed::channelClosed(), which says nothing about why. + self::expectException(ChannelException::class); + self::expectExceptionMessage(self::CHANNEL_CLOSED); + + (new ConsumerRunner($connection))->run($consumer); + } + + private function givenConnectionTo(Channel $channel): Connection + { + $connection = self::createStub(Connection::class); + // Like BunnyConnection: a Fiber of its own, so that the runner's own await() suspends that + // Fiber rather than the main context - the one place React's scheduler Fiber is shared with + // whatever else ran before in this process. + $connection->method('run')->willReturnCallback( + static fn (Closure $operation) => await(async($operation)()), + ); + $connection->method('getChannel')->willReturn($channel); + + return $connection; + } +} diff --git a/tests/ConsumerRunnerTest.php b/tests/ConsumerRunnerTest.php index 84181fe..7b99718 100644 --- a/tests/ConsumerRunnerTest.php +++ b/tests/ConsumerRunnerTest.php @@ -14,9 +14,11 @@ use Cdn77\RabbitMQBundle\RabbitMQ\Operation\AcknowledgeOperation; use Cdn77\RabbitMQBundle\RabbitMQ\Queue; use Cdn77\RabbitMQBundle\Tests\RabbitMQ\InMemoryConsumer; +use Cdn77\RabbitMQBundle\Tests\RabbitMQ\ThrowingConsumer; use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\Attributes\Group; use PHPUnit\Framework\TestCase; +use RuntimeException; use function end; @@ -41,12 +43,69 @@ public function setUp(): void public function tearDown(): void { $this->clearRabbitMQ(); + $this->getConnection()->disconnect(); parent::tearDown(); } #[DataProvider('maxMessagesDataProvider')] public function testMaxMessagesLimit(int $maxMessages): void + { + $queue = $this->givenQueueWithEnoughMessages(); + $consumer = $this->givenConfiguredConsumer($maxMessages, $queue); + + $this->whenConsume($consumer); + + $this->thenOnlyMaxMessagesCountIsConsumed($maxMessages, $consumer); + } + + public function testConsumerExceptionIsPropagated(): void + { + $queue = $this->givenQueueWithEnoughMessages(); + + // The consumer callback runs in its own Fiber, so an exception thrown there can only end + // up as an unhandled promise rejection unless the runner routes it out of run(). maxSeconds + // is a safety net: without it a regression would block the suite instead of failing it. + $consumer = new ThrowingConsumer(new Configuration($queue->getName(), 1, 0, null, 5.0)); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage(ThrowingConsumer::EXCEPTION_MESSAGE); + + $this->whenConsume($consumer); + } + + public function testConsumerIsNotCalledAgainAfterItFailed(): void + { + $queue = $this->givenQueueWithEnoughMessages(); + + // Prefetch more than one message, so Bunny has further deliveries buffered by the time the + // first one fails. + $consumer = new ThrowingConsumer(new Configuration($queue->getName(), 10, 0, null, 5.0)); + + try { + $this->whenConsume($consumer); + + self::fail('The consumer exception should have been propagated'); + } catch (RuntimeException $error) { + self::assertSame(ThrowingConsumer::EXCEPTION_MESSAGE, $error->getMessage()); + } + + // The failure settles the awaited promise only on a future tick, while Bunny hands over the + // next buffered delivery as soon as the callback returns - a consumer that has just failed + // must not be given those, they belong back in the queue. + self::assertSame(1, $consumer->getConsumeCallCount()); + } + + private function clearRabbitMQ(): void + { + $connection = $this->getConnection(); + $connection->run(static function () use ($connection): void { + $connection->getChannel()->queueDelete('testQueue'); + $connection->getChannel()->exchangeDelete('test'); + }); + } + + private function givenQueueWithEnoughMessages(): Queue { $exchange = new Exchange('test', new ExchangeType(ExchangeType::DIRECT)); $queue = new Queue('testQueue'); @@ -60,25 +119,19 @@ public function testMaxMessagesLimit(int $maxMessages): void $this->setupTopology($topology); $this->givenEnoughMessagesInQueue($exchange, $routingKey); - $consumer = $this->givenConfiguredConsumer($maxMessages, $queue); - - $this->whenConsume($consumer); - $this->thenOnlyMaxMessagesCountIsConsumed($maxMessages, $consumer); - } - - private function clearRabbitMQ(): void - { - $this->getConnection()->getChannel()->queueDelete('testQueue'); - $this->getConnection()->getChannel()->exchangeDelete('test'); + return $queue; } private function givenEnoughMessagesInQueue(Exchange $exchange, string $routingKey): void { - $channel = $this->getConnection()->getChannel(); - for ($i = 1; $i <= 10; $i++) { - $channel->publish((string) $i, [], $exchange->getName(), $routingKey); - } + $connection = $this->getConnection(); + $connection->run(static function () use ($connection, $exchange, $routingKey): void { + $channel = $connection->getChannel(); + for ($i = 1; $i <= 10; $i++) { + $channel->publish((string) $i, [], $exchange->getName(), $routingKey); + } + }); } private function givenConfiguredConsumer(int $maxMessages, Queue $queue): InMemoryConsumer diff --git a/tests/EventListener/DisconnectConnectionTest.php b/tests/EventListener/DisconnectConnectionTest.php new file mode 100644 index 0000000..c0b62ce --- /dev/null +++ b/tests/EventListener/DisconnectConnectionTest.php @@ -0,0 +1,95 @@ +createMock(Connection::class); + $connection->expects(self::exactly(2))->method('disconnect'); + + $dispatcher = $this->givenDispatcher($connection); + + // Only the absence of an interrupting signal on the event matters here, and the real + // TerminateEvent is final and would drag symfony/http-foundation in just to be built. + $dispatcher->dispatch(new stdClass(), KernelEvents::TERMINATE); + $dispatcher->dispatch($this->consoleTermination(null), ConsoleEvents::TERMINATE); + } + + /** + * In its own process: the subscriber stops the event loop, which is process-wide and outlives + * the test - React's shared scheduler Fiber is left in a run() that returns the moment anything + * resumes it, so the next await() in the suite would die with + * `AssertionError: assert(\is_callable($ret))`. + */ + #[RunInSeparateProcess] + #[PreserveGlobalState(false)] + public function testDoesNotDisconnectWhenTerminatingFromSignalHandler(): void + { + $connection = $this->createMock(Connection::class); + $connection->expects(self::never())->method('disconnect'); + + $dispatcher = $this->givenDispatcher($connection); + + $dispatcher->dispatch($this->consoleTermination(self::SIGNAL), ConsoleEvents::TERMINATE); + } + + public function testDisconnectsFromInsideFiber(): void + { + $connection = $this->createMock(Connection::class); + $connection->expects(self::once())->method('disconnect'); + + $dispatcher = $this->givenDispatcher($connection); + + // Being on a Fiber says nothing about being in a signal handler: a command that ended up + // dispatching its termination from one still has to have its heartbeat timer taken off the + // event loop, or the process would never exit. + $event = $this->consoleTermination(null); + $fiber = new Fiber(static function () use ($dispatcher, $event): void { + $dispatcher->dispatch($event, ConsoleEvents::TERMINATE); + }); + $fiber->start(); + + self::assertTrue($fiber->isTerminated()); + } + + private function givenDispatcher(Connection $connection): EventDispatcher + { + $dispatcher = new EventDispatcher(); + $dispatcher->addSubscriber(new DisconnectConnection($connection)); + + return $dispatcher; + } + + private function consoleTermination(int|null $interruptingSignal): ConsoleTerminateEvent + { + return new ConsoleTerminateEvent( + new Command('test'), + new ArrayInput([]), + new NullOutput(), + 0, + $interruptingSignal, + ); + } +} diff --git a/tests/RabbitMQ/BunnyConnectionTest.php b/tests/RabbitMQ/BunnyConnectionTest.php new file mode 100644 index 0000000..5da8f5b --- /dev/null +++ b/tests/RabbitMQ/BunnyConnectionTest.php @@ -0,0 +1,92 @@ +getConnection(); + + $stale = $connection->run(static fn () => $connection->getChannel()); + + // RabbitMQ answers a publish to an exchange that does not exist by closing the channel. + $connection->run( + static fn () => $connection->getChannel()->publish('body', [], 'noSuchExchange', 'aKey'), + ); + + // That channel.close arrives on its own, with nothing awaiting it, so give the loop a turn. + $connection->run(static fn () => delay(self::CLOSE_ARRIVAL)); + + $fresh = $connection->run(static fn () => $connection->getChannel()); + + // A working channel, not just a different object: the broker answers on it. + $declareOk = $connection->run( + static fn () => $connection->getChannel()->queueDeclare(self::QUEUE), + ); + $connection->run(static fn () => $connection->getChannel()->queueDelete(self::QUEUE)); + + self::assertNotSame($stale, $fresh); + self::assertSame(self::QUEUE, $declareOk->queue); + } + + /** + * The Fiber the operation gets is part of the contract: reusing the calling one - as an + * acknowledge issued from inside a consumer callback would - leaves PHP unable to switch back + * out of a context that forbids it, and a shutdown from a signal handler dies with a FiberError + * instead of just skipping the disconnect. run() itself talks to nobody, so no broker needed. + */ + public function testRunGivesEveryOperationAFiberOfItsOwn(): void + { + $connection = new BunnyConnection( + ConnectionConfiguration::fromDsn(new Dsn('amqp://127.0.0.1/')), + ); + + $callingFiber = Fiber::getCurrent(); + $outer = null; + $inner = null; + + $connection->run(static function () use ($connection, &$outer, &$inner): void { + $outer = Fiber::getCurrent(); + + $connection->run(static function () use (&$inner): void { + $inner = Fiber::getCurrent(); + }); + }); + + self::assertInstanceOf(Fiber::class, $outer); + self::assertInstanceOf(Fiber::class, $inner); + self::assertNotSame($callingFiber, $outer); + self::assertNotSame($outer, $inner); + } + + protected function tearDown(): void + { + $this->getConnection()->disconnect(); + } +} diff --git a/tests/RabbitMQ/ConfigurationTest.php b/tests/RabbitMQ/ConfigurationTest.php index f46c290..09db065 100644 --- a/tests/RabbitMQ/ConfigurationTest.php +++ b/tests/RabbitMQ/ConfigurationTest.php @@ -28,7 +28,6 @@ public function testConnectionConfiguration(): void self::assertSame('password', $configuration->getPassword()); self::assertSame(60, $configuration->getHeartbeat()); self::assertSame(10, $configuration->getConnectionTimeout()); - self::assertSame(11, $configuration->getReadWriteTimeout()); } public function testTopologyConfiguration(): void diff --git a/tests/RabbitMQ/ConfigurationTest.yaml b/tests/RabbitMQ/ConfigurationTest.yaml index 1b421cf..f6e2287 100644 --- a/tests/RabbitMQ/ConfigurationTest.yaml +++ b/tests/RabbitMQ/ConfigurationTest.yaml @@ -1,5 +1,5 @@ rabbitmq: - dsn: amqp://guest:password@127.0.0.1:5672/?heartbeat=60&connection_timeout=10&read_write_timeout=11 + dsn: amqp://guest:password@127.0.0.1:5672/?heartbeat=60&connection_timeout=10 exchanges: exchange1: diff --git a/tests/RabbitMQ/ThrowingConsumer.php b/tests/RabbitMQ/ThrowingConsumer.php new file mode 100644 index 0000000..8ad326c --- /dev/null +++ b/tests/RabbitMQ/ThrowingConsumer.php @@ -0,0 +1,48 @@ +configuration = $configuration; + } + + public function consume(Message $message): void + { + $this->consumeCallCount++; + + throw new RuntimeException(self::EXCEPTION_MESSAGE); + } + + public function getConsumeCallCount(): int + { + return $this->consumeCallCount; + } + + public function getName(): string + { + return 'throwing'; + } + + public function getConfiguration(): Configuration + { + return $this->configuration; + } +} From 440090233ded5139dbe749d9b26e823934d20cac Mon Sep 17 00:00:00 2001 From: Pavel Vondrak Date: Tue, 25 Aug 2026 16:05:13 +0200 Subject: [PATCH 2/2] fix!: bound broker operations and replace stale connections A producer process could stop making progress for as long as it ran. Publishing goes through Connection::run(), which awaits on the event loop and suspends it again the moment the operation settles - so between two publishes nothing drives the loop, no heartbeat frame can be written, and a broker with heartbeat=60 hangs up on a command that spends minutes on work of its own. A firewall silently blackholing an idle socket looks the same from here. The next operation then awaited a reply that could never arrive. Bunny settles every protocol wait from an incoming frame and rejects none of them when the socket dies (Connection::$awaitList), and it reports connection-level failures as an 'error' event that nobody listened to, so the awaited promise simply stayed pending. A sync command was observed stuck in hrtimer_nanosleep for 20 hours. Even a socket the broker had closed cleanly only got as far as a fatal AssertionError from React's scheduler, one operation later. Connection::run() now races the operation against the client's 'error' event and against a timer of operation_timeout seconds, and throws OperationFailed rather than waiting forever. The client is discarded on either, because a timed-out operation is still parked inside the loop and an error means the connection is gone. ConsumerRunner is the one caller that must not be bounded - its loop runs until the consumer's own message or time limit - so it takes the new runWithoutTimeout(). Not everything that comes out of that await is a failure of the connection, though. An exception from the caller's own closure leaves by the same door, and Bunny re-emits every channel error onto the client (Client::channel()), so a client 'error' is no proof either. Discarding on those cost a reconnect and whatever was still unflushed elsewhere on the connection - measured with rabbitmqctl list_connections, a 404 publish took the connection from 2 to 0. So the client is now replaced on a timeout, where the operation is parked inside the loop for good, or when it can no longer be disconnected, which is the state Bunny leaves it in when it tears the client down itself for a connection that is gone. A channel the broker closed is replaced on its own 'close' event, as before. Only the wait for messages is exempt there. Getting to it opens a channel and sends basic.qos and basic.consume, each of which waits for a reply frame and so would hang for good on a socket that dies mid-handshake; the runner's own watch for channel failures is no help however early it is installed, since a Fiber stuck in the handshake never reaches the await() it rejects. That startup therefore runs inside a bounded run() of its own. A prefetch the broker refuses still arrives as ConfigurationFailed: Bunny's await list rejects the pending basic.qos-ok before that frame reaches the channel and the client, so the operation's own catch gets there first, and an integration test pins that order. SetupAction goes the other way and takes a run() per declaration rather than one around the whole topology. operation_timeout is what a single round trip may take, and a topology of hundreds of items would otherwise run out of it while every one of them was answered promptly. It also puts the timeout where it can be reported: run() raises it from outside the operation's Fiber, which stays suspended in the frame it is waiting for, so the per-item catch never saw it and setup() leaked OperationFailed instead of naming the exchange or queue that hung. GetOperation is split the same way, for the first of those reasons: a read asks for as many basic.get round trips as it wants messages, and one budget shared between them ends a read that was answered promptly throughout - 5000 gets, each about a tenth of a millisecond, ran out of a 0.5s bound - taking the connection with it. The timer and listener a run() per get adds cost 0.52s against 0.83s for 2000 gets locally, which is noise beside any real network round trip. Since the loop is frozen between operations, a connection idle for longer than the heartbeat it promised cannot have sent one and is assumed closed: run() replaces it up front instead of publishing into a hole. Whether a call is nested - and so already driving the loop, in which case the connection stays - is counted rather than read off Fiber::getCurrent(): that stands for "inside an operation of ours" only until the application brings Fibers of its own, as anything built on React does, and such a caller would have had the check skipped for the rest of the process. The teardown behind all of this closes the client locally (RAW_CONNECTION_INACTIVE) rather than exchanging connection.close with a broker that may be gone - which is also the only path that reaches Bunny's Connection::disconnect(), the one place that cancels the heartbeat timer that would otherwise keep the process alive with no stream left to wake it. It is the fallback on every failure path, and what the console listener uses before stopping the loop; kernel.terminate must not stop it, as a php-fpm worker serves further requests in the same process. That stop is not the belt and braces it looks like. One socket the teardown cannot reach is the one Bunny abandons: Client::connect() neither closes the connection nor rolls the state back when the handshake throws, so the client stays Connecting, refuses to be disconnected, and its socket is left registered on the event loop. A process holding one cannot exit, since React's shutdown blocks in stream_select() with nothing to wake it - a command that could not reach the broker ran until it was killed 25 seconds later, and exits at once with the loop stopped. Closing that socket belongs upstream, in that catch; stopping the loop is what a terminating command can do about it from here, at the price of dropping work a later console.terminate listener put on the loop without awaiting it. A subprocess test holds that line in place, since a process that hangs cannot be told from one that exits from the inside. disconnect() itself asks for the connection.close handshake first, and falls back to that local teardown. Not for the courtesy: a publish() sits in React's write buffer until the loop turns again, and closing the stream locally reaches React's close(), which discards the buffer rather than flushing it like end(). Every producer that published and then let kernel.terminate close the connection therefore lost its message and left "client unexpectedly closed TCP connection" in the broker log - the same for an acknowledge issued as the last thing a consumer does. The handshake awaits a reply, so the loop turns and the buffer goes out ahead of it - belt to the brace of the flush below, which has since made every path write before disconnect() is even reached. It stays because it is the proper close, the one that leaves no such line behind. Bounded by run() and skipped for a stale connection, so a broker that has gone away still cannot hold a terminating process: with the broker paused, disconnect() returns after exactly operation_timeout and the process exits. A heartbeat of zero is refused outright. Bunny arms the heartbeat timer whatever the interval is and re-arms it with the same value, so 0 - the AMQP way of switching heartbeats off - leaves a timer that is due again the moment it fires: it spins the event loop at a full core for the whole of every operation and floods the broker with heartbeat frames (0.78s of CPU for a one-second await, against 0.00s at 60). The guard sits in Configuration\Connection, which is where the DSN parameter, the container key and a hand-built configuration all meet - including the blind cast that turns any non-numeric value into a zero. A run ends between messages, never in the middle of one. A handler that awaits anything - a batch publish committing, a get, any round trip - suspends its own Fiber and turns the loop, which is exactly where the maxSeconds timer falls due: ending the run there had run() return, the consumer cancelled and the connection closed by kernel.terminate or console.terminate while that handler was still parked. It then resumed into a connection that was gone, so its acknowledge was lost and the message redelivered - and had it thrown, the rejection landed on a promise settled long ago, which react/promise drops, so the command exited successfully having half-processed a message. Measured: the runner returned at 0.34s, the handler acknowledged at 1.05s, the message went back on the queue, and the process then hung because that late acknowledge reconnected and armed a fresh heartbeat timer. The stop is therefore recorded while a message is being handled and the delivery callback settles it on its way out; only one is ever in flight, since Bunny queues deliveries and runs them with a concurrency of 1. A failure is not held back the same way - it comes from a channel that errored or closed, and a handler parked on a dead socket never resumes at all, so waiting for it would block the run for good. A consumer whose consume() blocks for longer than the heartbeat loses its connection, and this cannot fix that: the loop only turns while something awaits it, and a delivery callback is synchronous PHP. The broker hangs up after two missed intervals, and an acknowledge issued after that reports success into a dead socket - Channel::ack() awaits no reply - so the message is redelivered and the handler runs again. Reconnecting for it would be worse, since delivery tags belong to the channel that is gone. What is left is to say so: the fallback the runner reports it through no longer claims the broker closed the channel, and docs/Consuming.md spells out that such a handler needs a heartbeat above twice its worst case, or must stop blocking the loop. What the handler published before blocking is not lost with the rest, at least - see the flush below. A batch reports its failure as OperationFailed however that failure arrives, so handleAll() wraps around run() rather than inside the operation. The failure does not always come back through the operation's Fiber: the await list resumes the commit with the broker's channel.close, so the catch does run, but the best-effort txRollback() then awaits a reply of its own - and that suspension hands the read loop the very frame that caused all this. It reaches the channel, whose 'error' the client re-emits, and run() loses the race while the operation is still parked in the rollback, leaving the OperationFailed after it in an orphaned promise. A batch published to a missing exchange leaked Bunny's ChannelException. Exceptions of the bundle's own are rethrown unwrapped there, so a connection that could not be opened and an operation the timeout ended keep saying which they were. A fire-and-forget write is on the socket before the operation returns. React's stream buffers what it is handed and writes it when a loop iteration finds the socket writable, and publish(), ack() and nack() await no reply, so an operation made of them alone never suspends and no iteration happens - Bunny asks for a drain of its own only once that buffer passes React's 64 KiB soft limit, which takes a body of 65488 bytes with the smallest framing there is. Below that a producer of ordinary messages sent nothing at all: twelve publishes half a second apart reached the broker as none of them, and at heartbeat=1 the broker hung up on a producer that had been publishing throughout - while lastOperationAt was refreshed by every one of those publishes, so the connection never counted as stale either. Every operation now ends by giving the loop that one iteration: a timer whose callback asks for a future tick, since an await settled by either alone resumes from inside the loop's own tick or timer phase, before the stream_select() that does the writing. Twelve of twelve, at about 7us an operation. Nested calls included, which the consume loop needs both ways. A handler that publishes and then works on for a minute holds the loop for that minute, so its message sat unwritten and went with the connection if that was lost first - measured, nothing at the broker for the whole of an 8s handler and the message there the moment it returned. And once the wait for messages is over nothing turns the loop again, which is where a consumer's last acknowledge is written, so runWithoutTimeout() flushes on its way out too. About 10us a nested operation, and the two writes of one message are no longer atomic as a result: a connection lost mid-handler leaves the published message standing while the consumed one is redelivered, which is at-least-once with a duplicate rather than a message that disappears without a word. Both cases are covered by tests that publish in a subprocess and kill it - letting it exit would run React's shutdown, which turns the loop and flushes the buffer whether the code under test did or not. BREAKING CHANGE: * RabbitMQ\Connection gained runWithoutTimeout(Closure): mixed, so custom implementations need updating. Only the consume loop should use it; everything else wants the bounded run(). * Operations that used to wait indefinitely now throw Exception\OperationFailed after operation_timeout seconds (default 30, and configurable per DSN parameter or YAML key). Any value of 0 or below restores the old behaviour. Callers that report failures in their own terms keep doing so with it as the cause: Exception\ConnectionFailed for connecting, Exception\ConfigurationFailed for topology setup. * heartbeat has to be positive now. Zero, which in AMQP switches heartbeats off, throws Exception\ConfigurationFailed - Bunny 0.6 cannot turn them off, and spins the event loop instead. Configure a long interval for as few heartbeats as possible. --- docs/Consuming.md | 24 ++ docs/Producing.md | 8 + docs/Setup.md | 24 +- src/Configuration/Connection.php | 51 ++- src/ConsumerRunner.php | 210 ++++++++---- src/DependencyInjection/Configuration.php | 6 + src/EventListener/DisconnectConnection.php | 18 ++ src/Exception/ConfigurationFailed.php | 5 + src/Exception/ConnectionFailed.php | 13 +- src/Exception/OperationFailed.php | 13 + src/RabbitMQ/BunnyConnection.php | 300 ++++++++++++++++-- src/RabbitMQ/Connection.php | 15 + src/RabbitMQ/Operation/GetOperation.php | 33 +- src/RabbitMQ/Operation/PublishOperation.php | 79 +++-- src/SetupAction.php | 102 +++--- tests/Configuration/ConnectionTest.php | 41 +++ tests/ConsumerRunnerFailureTest.php | 66 +++- tests/ConsumerRunnerTest.php | 264 ++++++++++++++- .../DisconnectConnectionTest.php | 92 +++++- tests/RabbitMQ/BunnyConnectionTest.php | 256 ++++++++++++++- tests/RabbitMQ/Operation/GetOperationTest.php | 84 +++++ .../Operation/PublishOperationTest.php | 149 +++++++++ tests/RabbitMQ/SuspendingConsumer.php | 68 ++++ tests/SetupActionTest.php | 61 ++++ tests/fixtures/console-termination.php | 43 +++ .../publish-from-a-blocking-handler.php | 64 ++++ tests/fixtures/publish-without-flushing.php | 32 ++ 27 files changed, 1914 insertions(+), 207 deletions(-) create mode 100644 tests/Configuration/ConnectionTest.php create mode 100644 tests/RabbitMQ/Operation/GetOperationTest.php create mode 100644 tests/RabbitMQ/Operation/PublishOperationTest.php create mode 100644 tests/RabbitMQ/SuspendingConsumer.php create mode 100644 tests/SetupActionTest.php create mode 100644 tests/fixtures/console-termination.php create mode 100644 tests/fixtures/publish-from-a-blocking-handler.php create mode 100644 tests/fixtures/publish-without-flushing.php diff --git a/docs/Consuming.md b/docs/Consuming.md index 9419bdf..856733e 100644 --- a/docs/Consuming.md +++ b/docs/Consuming.md @@ -65,4 +65,28 @@ Consumer configuration is done via `getConfiguration()` that returns instance of `maxMessages` & `maxSeconds` parameters are recommended to set to something else than `null` so consumer will shutdown gracefully after specified number of messages is consumed or seconds elapsed so it can start with fresh memory again. +Both limits end the run between messages, never in the middle of one: a `consume()` that is still +running when `maxSeconds` falls due is left to finish, and its acknowledge goes out on a connection +that is still open. + Consumer is registered under the name specified in `getName()` method. You can check whether it is successfully registered through `debug:rabbitmq:consumers` command. It can be run with `rabbitmq:consumer:run example_consumer` + +> **Heads up (Bunny 0.6): a `consume()` that blocks for longer than the heartbeat loses the +> connection.** The event loop only turns while something awaits it, and a handler is ordinary +> synchronous PHP - a subprocess, an SFTP upload, a long HTTP call - so no heartbeat frame can be +> written for as long as it runs. The broker hangs up after two missed intervals, and the consumer +> then dies with `Exception\ConnectionFailed` saying the channel closed with no error reported. +> +> Worse than the exception: an acknowledge issued *after* that point is a write with no reply to +> wait for, so it silently goes nowhere. The message stays unacknowledged, the broker requeues it, +> and the handler runs again on the next consumer - work you may have thought was committed. +> +> A message *published* from inside a handler does not share that fate: the bundle puts it on the +> socket before `handle()` returns, so a handler that publishes and then blocks still gets it out. +> The two are therefore not atomic - if the connection goes while the handler works on, the +> published message stands while the consumed one is requeued and handled again. +> +> So a handler that can take minutes needs a `heartbeat` comfortably above twice its worst case +> (`heartbeat: 3600` for handlers measured in minutes), or it needs to stop blocking the loop. +> Acknowledging *before* the long stretch, rather than after it, also keeps the acknowledge on a +> connection that is still alive - at the cost of at-most-once delivery for that message. diff --git a/docs/Producing.md b/docs/Producing.md index e1144b9..1a12dc1 100644 --- a/docs/Producing.md +++ b/docs/Producing.md @@ -48,3 +48,11 @@ final class ExampleProducer > command returns), so HTTP and console producers are handled automatically. If you publish from a > context that dispatches neither event (e.g. a bare script using the event loop directly), call > `$connection->disconnect()` yourself when you're done. + +> **Heads up (Bunny 0.6):** a single `handle()` is fire-and-forget - the broker sends no reply to +> wait for - so it tells you the message was written to the socket, not that the broker has it. The +> bundle does make sure of the writing: publishing awaits nothing, so the event loop would otherwise +> never turn and the message would sit in a buffer until some later operation happened to flush it, +> which for a producer of ordinary messages could be never. Use `handleAll()` when you need to know +> the broker has the messages before you carry on: it publishes in a transaction and returns once +> the commit is confirmed. diff --git a/docs/Setup.md b/docs/Setup.md index b163008..bb2ee63 100644 --- a/docs/Setup.md +++ b/docs/Setup.md @@ -4,13 +4,35 @@ This bundle uses `rabbitmq` container extension key and is able to merge configu These configuration options are available to setup connection to your RabbitMQ instance. -Example DSN: `amqp://username:password@host:1234/vhost?heartbeat=60&connection_timeout=10` +Example DSN: `amqp://username:password@host:1234/vhost?heartbeat=60&connection_timeout=10&operation_timeout=30` ```yaml rabbitmq: dsn: '%env(RABBITMQ_DSN)%' + heartbeat: 60 # seconds; also as a DSN parameter, where it wins over this + connection_timeout: 10 # seconds to establish the connection + operation_timeout: 30 # seconds a single broker operation may take before it is given up on ``` +`operation_timeout` bounds every single operation - publishing, acknowledging, declaring topology, +connecting - but never the consume loop, which runs until the consumer's own message or time limit +is reached. Exceeding it replaces the connection and throws `Exception\OperationFailed`, rather than +leaving the process waiting for a broker that is not going to answer. Where the caller reports +failures in its own terms it keeps doing so, carrying the `OperationFailed` as the cause: connecting +throws `Exception\ConnectionFailed`, and `rabbitmq:setup` throws `Exception\ConfigurationFailed` +naming the exchange or queue it got stuck on. All of them implement `Exception\Exception`. Any value +of `0` or below disables the bound, at the risk of a process that can never finish. + +A connection idle for longer than `heartbeat` is replaced before the next operation: the event loop +only turns while an operation is awaiting, so nothing can send a heartbeat frame in between, and the +broker will have closed such a connection already. + +`heartbeat` has to be a positive number of seconds; `0`, which in AMQP switches heartbeats off, is +rejected. Bunny 0.6 arms the timer whatever the interval is and re-arms it with the same value, so +zero leaves a timer that is due again the moment it fires - spinning the event loop at a full core +for as long as any operation is awaiting, and flooding the broker with heartbeat frames. Configure a +long interval (say `3600`) if heartbeats are genuinely not wanted. + Exchanges and Queues configuration can be done this way ```yaml diff --git a/src/Configuration/Connection.php b/src/Configuration/Connection.php index 8ab1c96..f67ca6b 100644 --- a/src/Configuration/Connection.php +++ b/src/Configuration/Connection.php @@ -5,11 +5,15 @@ namespace Cdn77\RabbitMQBundle\Configuration; use Cdn77\RabbitMQBundle\DependencyInjection\Configuration; +use Cdn77\RabbitMQBundle\Exception\ConfigurationFailed; + +use function is_numeric; final class Connection { private const int DEFAULT_HEARTBEAT = 60; private const int DEFAULT_CONNECTION_TIMEOUT = 3; + private const float DEFAULT_OPERATION_TIMEOUT = 30.0; /** @var string */ private $host; @@ -32,6 +36,9 @@ final class Connection /** @var int */ private $connectionTimeout; + /** @var float */ + private $operationTimeout; + public function __construct( string $host, int $port, @@ -40,14 +47,16 @@ public function __construct( string|null $password, int $heartbeat = self::DEFAULT_HEARTBEAT, int $connectionTimeout = self::DEFAULT_CONNECTION_TIMEOUT, + float $operationTimeout = self::DEFAULT_OPERATION_TIMEOUT, ) { $this->host = $host; $this->port = $port; $this->vhost = $vhost; $this->user = $user; $this->password = $password; - $this->heartbeat = $heartbeat; + $this->heartbeat = self::validHeartbeat($heartbeat); $this->connectionTimeout = $connectionTimeout; + $this->operationTimeout = $operationTimeout; } /** @param mixed[] $configuration */ @@ -60,7 +69,9 @@ public static function fromDI(array $configuration): self isset($configuration[Configuration::KEY_CONFIGURATION_HEARTBEAT]) && ! isset($dsn->getParameters()[Configuration::KEY_CONFIGURATION_HEARTBEAT]) ) { - $new->heartbeat = (int) $configuration[Configuration::KEY_CONFIGURATION_HEARTBEAT]; + $new->heartbeat = self::validHeartbeat( + (int) $configuration[Configuration::KEY_CONFIGURATION_HEARTBEAT], + ); } if ( @@ -70,12 +81,23 @@ public static function fromDI(array $configuration): self $new->connectionTimeout = (int) $configuration[Configuration::KEY_CONFIGURATION_CONNECTION_TIMEOUT]; } + // Only a number, never a blind cast: garbage would become 0.0, which is how the timeout + // is switched off - the one value nobody means to configure by accident. + $operationTimeout = $configuration[Configuration::KEY_CONFIGURATION_OPERATION_TIMEOUT] ?? null; + if ( + is_numeric($operationTimeout) + && ! isset($dsn->getParameters()[Configuration::KEY_CONFIGURATION_OPERATION_TIMEOUT]) + ) { + $new->operationTimeout = (float) $operationTimeout; + } + return $new; } public static function fromDsn(Dsn $dsn): self { $parameters = $dsn->getParameters(); + $operationTimeout = $parameters[Configuration::KEY_CONFIGURATION_OPERATION_TIMEOUT] ?? null; return new self( $dsn->getHost(), @@ -86,6 +108,7 @@ public static function fromDsn(Dsn $dsn): self (int) ($parameters[Configuration::KEY_CONFIGURATION_HEARTBEAT] ?? self::DEFAULT_HEARTBEAT), (int) ($parameters[Configuration::KEY_CONFIGURATION_CONNECTION_TIMEOUT] ?? self::DEFAULT_CONNECTION_TIMEOUT), + is_numeric($operationTimeout) ? (float) $operationTimeout : self::DEFAULT_OPERATION_TIMEOUT, ); } @@ -123,4 +146,28 @@ public function getConnectionTimeout(): int { return $this->connectionTimeout; } + + /** How long a single broker operation may take before it is given up on. Zero disables it. */ + public function getOperationTimeout(): float + { + return $this->operationTimeout; + } + + /** + * Bunny 0.6.0-alpha.4 arms a heartbeat timer whatever the interval is and re-arms it with the + * same value, so 0 - the AMQP way of switching heartbeats off - leaves a timer that is due + * again the moment it fires. It then spins the event loop at a full core for the whole of every + * operation and floods the broker with heartbeat frames: measured at 0.78s of CPU for a + * one-second await, against 0.00s with an interval of 60. Rejected here rather than in + * BunnyConnection, so that a DSN parameter, a container key and a hand-built configuration are + * all covered - including the blind cast above, which turns any non-numeric value into a zero. + */ + private static function validHeartbeat(int $heartbeat): int + { + if ($heartbeat > 0) { + return $heartbeat; + } + + throw ConfigurationFailed::heartbeatMustBePositive($heartbeat); + } } diff --git a/src/ConsumerRunner.php b/src/ConsumerRunner.php index 132524a..e1f86e1 100644 --- a/src/ConsumerRunner.php +++ b/src/ConsumerRunner.php @@ -4,6 +4,7 @@ namespace Cdn77\RabbitMQBundle; +use Bunny\ChannelInterface; use Bunny\Message; use Cdn77\RabbitMQBundle\Exception\ConfigurationFailed; use Cdn77\RabbitMQBundle\Exception\ConnectionFailed; @@ -38,104 +39,173 @@ public function run(Consumer $consumer): void return; } - $this->connection->run(function () use ($consumer, $configuration): void { - $channel = $this->connection->getChannel(); - - try { - $channel->qos($configuration->getPrefetchSize(), $configuration->getPrefetchCount()); - } catch (Throwable $error) { - throw ConfigurationFailed::invalidPrefetchValues($error); - } - + // The one caller that must not be bounded by the operation timeout: waiting for messages + // ends when the consumer's message or time limit is reached, which is minutes or hours from + // here. Getting there is bounded all the same - see the startup below. + $this->connection->runWithoutTimeout(function () use ($consumer, $configuration): void { /** @var Deferred $stopped */ $stopped = new Deferred(); $stopping = false; + // Written by the delivery callback and read by the closures below, which phpstan cannot + // see through a by-reference binding: without the annotations it narrows both to false. + /** @var bool $handling */ + $handling = false; + /** @var bool $stopWhenHandled */ + $stopWhenHandled = false; // Settle on a future tick, never straight from a delivery callback: doing it inline // would resume the Fiber awaiting below from inside the callback's own Fiber, leaving // React's scheduler with no way to hand the result back to run()'s caller. The flag has // to be set right away though: Bunny hands over the next buffered delivery as soon as // the callback returns, long before a future tick gets to run. - $stop = static function () use (&$stopping, $stopped): void { + // + // And not while a message is being handled. A handler that awaits anything - a publish + // that commits, a get, any round trip - suspends its own Fiber and turns the loop, which + // is where the time limit falls due: ending the run there would have run() return, the + // consumer cancelled and the connection closed by kernel/console.terminate while that + // handler was still parked. It then resumes into a connection that is gone, so its + // acknowledge is lost and the message redelivered, and should it throw, the rejection + // lands on a promise that settled long ago and is dropped - the command exiting + // successfully having half-processed a message. The last thing the handler does is + // settle this instead. Only ever one of them: Bunny queues deliveries and runs them + // with a concurrency of 1. + $stop = static function () use (&$stopping, &$handling, &$stopWhenHandled, $stopped): void { $stopping = true; + if ($handling) { + $stopWhenHandled = true; + + return; + } + Loop::futureTick(static fn () => $stopped->resolve(null)); }; + // A failure is not held back for the handler, unlike the stop above: it comes from a + // channel that errored or closed, and a handler parked in a round trip on a connection + // that is gone never resumes at all - Bunny settles a protocol wait from an incoming + // frame and rejects none of them when the socket dies. Waiting for it would block the + // run for good, which is the failure this whole class is here to avoid. $fail = static function (Throwable $error) use (&$stopping, $stopped): void { $stopping = true; Loop::futureTick(static fn () => $stopped->reject($error)); }; + // A broker-closed channel emits both events: 'close' first, then 'error' carrying a + // ChannelException with the reply code and text. Only the first rejection counts, so + // hold this fallback back by a tick and let the one that says why go first. Nothing + // else settles the promise when the channel goes without an error - a lost connection + // closes it silently - so the fallback still gets there. + $closed = static fn () => Loop::futureTick( + static fn () => $fail(ConnectionFailed::channelClosedWithoutAnError()), + ); - $consumeOk = $channel->consume( - async(function (Message $message) use ( + // Bounded, unlike the wait for messages that follows it: opening the channel, + // basic.qos and basic.consume each wait for the broker to answer, and Bunny leaves + // such a wait pending for good when the socket dies - only an incoming frame ever + // settles one. The handlers below are no help here however early they are installed: + // a Fiber stuck in the handshake never reaches the await() they reject. + [$channel, $consumerTag] = $this->connection->run( + /** @return array{ChannelInterface, string} */ + function () use ( $consumer, $configuration, - $channel, $stop, $fail, + $closed, + $stopped, &$stopping, - ): void { + &$handling, + &$stopWhenHandled, + ): array { + $channel = $this->connection->getChannel(); + try { - // Further messages may already be in flight (prefetch) once the run is - // over - be it the limit being reached or the consumer having failed. - // Reject the ones we still get handed rather than only skipping them, so - // they go back to the queue right away instead of sitting unacknowledged - - // and invisible to other consumers - until the connection goes away. - // Whatever Bunny has buffered but not yet delivered is dropped by - // basic.cancel below and only the broker can put those back, which it does - // once this channel or connection closes. - if ( - $stopping - || ! $this->hasAnyMessageLeft( - $configuration->getMaxMessages(), - $this->processedMessageCount, - ) - ) { - $channel->nack($message, false, true); - - return; - } - - $consumer->consume($message); - - $this->processedMessageCount++; - - if ( - $this->hasAnyMessageLeft( - $configuration->getMaxMessages(), - $this->processedMessageCount, - ) - ) { - return; - } - - $stop(); + $channel->qos($configuration->getPrefetchSize(), $configuration->getPrefetchCount()); } catch (Throwable $error) { - // The callback runs in its own Fiber, so throwing here would only surface - // as an unhandled promise rejection. Hand the failure to the awaited - // promise instead to let it propagate out of run(). - $fail($error); + throw ConfigurationFailed::invalidPrefetchValues($error); } - }), - $configuration->getQueueName(), - ); - // Bunny reports channel and connection level failures as 'error'/'close' events rather - // than by throwing, and an unlistened event is silently dropped. Without these the - // await() below would keep blocking after e.g. the broker closed the channel or the - // connection was lost. Registered only now that consume() is through - it reports its - // own failures by throwing, and a rejection here would have nothing awaiting it yet. - // A broker-closed channel emits both events: 'close' first, then 'error' carrying a - // ChannelException with the reply code and text. Only the first rejection counts, so - // hold this fallback back by a tick and let the one that says why go first. Nothing - // else settles the promise when the channel goes without an error - a lost connection - // closes it silently - so the fallback still gets there. - $closed = static fn () => Loop::futureTick( - static fn () => $fail(ConnectionFailed::channelClosed()), + $consumeOk = $channel->consume( + async(function (Message $message) use ( + $consumer, + $configuration, + $channel, + $stop, + $fail, + $stopped, + &$stopping, + &$handling, + &$stopWhenHandled, + ): void { + $handling = true; + + try { + // Further messages may already be in flight (prefetch) once the + // run is over - be it the limit being reached or the consumer + // having failed. Reject the ones we still get handed rather than + // only skipping them, so they go back to the queue right away + // instead of sitting unacknowledged - and invisible to other + // consumers - until the connection goes away. Whatever Bunny has + // buffered but not yet delivered is dropped by basic.cancel below + // and only the broker can put those back, which it does once this + // channel or connection closes. + if ( + $stopping + || ! $this->hasAnyMessageLeft( + $configuration->getMaxMessages(), + $this->processedMessageCount, + ) + ) { + $channel->nack($message, false, true); + + return; + } + + $consumer->consume($message); + + $this->processedMessageCount++; + + if ( + $this->hasAnyMessageLeft( + $configuration->getMaxMessages(), + $this->processedMessageCount, + ) + ) { + return; + } + + $stop(); + } catch (Throwable $error) { + // The callback runs in its own Fiber, so throwing here would only + // surface as an unhandled promise rejection. Hand the failure to + // the awaited promise instead to let it propagate out of run(). + $fail($error); + } finally { + $handling = false; + + // The run was told to end while this handler had the loop: settling + // is its to do, now that it is finished with the message. A no-op + // if the failure above got there first. + if ($stopWhenHandled) { + Loop::futureTick(static fn () => $stopped->resolve(null)); + } + } + }), + $configuration->getQueueName(), + ); + + // Bunny reports channel and connection level failures as 'error'/'close' + // events rather than by throwing, and an unlistened event is silently dropped. + // Without these the await() below would keep blocking after e.g. the broker + // closed the channel or the connection was lost. Registered only now that + // consume() is through - it reports its own failures by throwing - and still + // in here, where no tick can pass between the two. + $channel->on('error', $fail); + $channel->once('close', $closed); + + return [$channel, $consumeOk->consumerTag]; + }, ); - $channel->on('error', $fail); - $channel->once('close', $closed); $maxSeconds = $configuration->getMaxSeconds(); $timer = $maxSeconds !== null ? Loop::addTimer($maxSeconds, $stop) : null; @@ -164,7 +234,7 @@ public function run(Consumer $consumer): void // for this consumer tag the moment cancel() returns, and messages the broker // still sends until it processes the frame are requeued when the channel or the // connection closes. - $channel->cancel($consumeOk->consumerTag, true); + $channel->cancel($consumerTag, true); } catch (Throwable $error) { // A channel the broker already closed can be neither used nor cancelled, and // saying so must not bury the failure that ended the run. diff --git a/src/DependencyInjection/Configuration.php b/src/DependencyInjection/Configuration.php index c46cdff..ea0d293 100644 --- a/src/DependencyInjection/Configuration.php +++ b/src/DependencyInjection/Configuration.php @@ -16,6 +16,7 @@ final class Configuration implements ConfigurationInterface public const string KEY_CONFIGURATION_DSN = 'dsn'; public const string KEY_CONFIGURATION_HEARTBEAT = 'heartbeat'; public const string KEY_CONFIGURATION_CONNECTION_TIMEOUT = 'connection_timeout'; + public const string KEY_CONFIGURATION_OPERATION_TIMEOUT = 'operation_timeout'; public const string KEY_CONFIGURATION_EXCHANGES = 'exchanges'; public const string KEY_CONFIGURATION_QUEUES = 'queues'; public const string KEY_EXCHANGE_NAME = 'name'; @@ -34,6 +35,7 @@ final class Configuration implements ConfigurationInterface private const string DEFAULT_DSN = 'amqp://127.0.0.1/'; private const int DEFAULT_HEARTBEAT = 60; private const int DEFAULT_TIMEOUT = 10; + private const int DEFAULT_OPERATION_TIMEOUT = 30; public function getConfigTreeBuilder(): TreeBuilder { @@ -61,6 +63,10 @@ private function configureConnection(ArrayNodeDefinition $rootNode): void $rootNode->children() ->scalarNode(self::KEY_CONFIGURATION_CONNECTION_TIMEOUT) ->defaultValue(self::DEFAULT_TIMEOUT); + + $rootNode->children() + ->scalarNode(self::KEY_CONFIGURATION_OPERATION_TIMEOUT) + ->defaultValue(self::DEFAULT_OPERATION_TIMEOUT); } private function configureExchanges(ArrayNodeDefinition $rootNode): void diff --git a/src/EventListener/DisconnectConnection.php b/src/EventListener/DisconnectConnection.php index 3c87595..63118d7 100644 --- a/src/EventListener/DisconnectConnection.php +++ b/src/EventListener/DisconnectConnection.php @@ -55,6 +55,24 @@ public function disconnect(object $event): void } $this->connection->disconnect(); + + if (! $event instanceof ConsoleTerminateEvent) { + return; + } + + // Not the belt and braces it looks like: a handshake that stalls leaves Bunny's socket on + // the loop for good - `Client::connect()` neither closes the connection nor rolls the state + // back when it throws, so `disconnect()` is left with a client it cannot disconnect - and + // React's shutdown then blocks in `stream_select()` with no timeout. A command that could + // not reach the broker would never exit. Both ways round in + // `DisconnectConnectionTest::testCommandExitsAfterAHandshakeThatStalled`. + // + // It costs what it says: process-wide and permanent, so a later `console.terminate` listener + // that puts work on the loop without awaiting it loses that work. Awaited work is fine - + // `await()` enters the loop itself - and a process that hangs is worse than either. Console + // only, all the same: a php-fpm worker serves further requests after `kernel.terminate`, and + // their first await() would resume a scheduler Fiber sitting in a loop that no longer runs. + Loop::stop(); } /** diff --git a/src/Exception/ConfigurationFailed.php b/src/Exception/ConfigurationFailed.php index 480f53d..3785076 100644 --- a/src/Exception/ConfigurationFailed.php +++ b/src/Exception/ConfigurationFailed.php @@ -14,6 +14,11 @@ final class ConfigurationFailed extends RuntimeException implements Exception { + public static function heartbeatMustBePositive(int $heartbeat): self + { + return new self(sprintf('Heartbeat must be a positive number of seconds, %d given', $heartbeat)); + } + public static function invalidPrefetchValues(Throwable|null $previous = null): self { return new self('Could not set prefetch-size/prefetch-count', 0, $previous); diff --git a/src/Exception/ConnectionFailed.php b/src/Exception/ConnectionFailed.php index f41e0a0..8564ff0 100644 --- a/src/Exception/ConnectionFailed.php +++ b/src/Exception/ConnectionFailed.php @@ -14,8 +14,17 @@ public static function causedBy(Throwable $previous): self return new self('Connection to RabbitMQ failed', 0, $previous); } - public static function channelClosed(): self + /** + * Said of a channel that closed without an error to explain it, which is not the broker + * refusing something on it - that comes with a reply code and text - but the channel going with + * the connection: lost, or torn down locally. + */ + public static function channelClosedWithoutAnError(): self { - return new self('Channel was closed by the broker'); + return new self( + 'Channel was closed with no error reported, so the connection was lost or closed locally' + . ' - a consumer handler that blocks the event loop for longer than the heartbeat is' + . ' the usual cause', + ); } } diff --git a/src/Exception/OperationFailed.php b/src/Exception/OperationFailed.php index 9feeddd..7e7bc59 100644 --- a/src/Exception/OperationFailed.php +++ b/src/Exception/OperationFailed.php @@ -5,7 +5,20 @@ namespace Cdn77\RabbitMQBundle\Exception; use RuntimeException; +use Throwable; + +use function sprintf; final class OperationFailed extends RuntimeException implements Exception { + public static function timedOut(float $seconds, Throwable|null $previous = null): self + { + return new self( + // %s, not a fixed number of decimals: a bound of 0.04 rounds to "0.0 seconds", which + // is the value that switches the bound off. + sprintf('Operation did not finish within %s seconds', $seconds), + 0, + $previous, + ); + } } diff --git a/src/RabbitMQ/BunnyConnection.php b/src/RabbitMQ/BunnyConnection.php index fa1451d..616e01a 100644 --- a/src/RabbitMQ/BunnyConnection.php +++ b/src/RabbitMQ/BunnyConnection.php @@ -6,22 +6,34 @@ use Bunny\ChannelInterface; use Bunny\Client; +use Bunny\ClientInterface; use Bunny\Configuration as BunnyConfiguration; use Bunny\Defaults; use Cdn77\RabbitMQBundle\Configuration; use Cdn77\RabbitMQBundle\Exception\CannotCreateChannel; use Cdn77\RabbitMQBundle\Exception\ConnectionFailed; +use Cdn77\RabbitMQBundle\Exception\OperationFailed; use Closure; +use React\EventLoop\Loop; +use React\Promise\Deferred; use Throwable; +use function microtime; use function React\Async\async; use function React\Async\await; +use function React\Promise\race; final class BunnyConnection implements Connection { /** @var BunnyConfiguration */ private $configuration; + /** @var float */ + private $operationTimeout; + + /** @var int */ + private $heartbeat; + /** @var Client */ private $client; @@ -31,6 +43,12 @@ final class BunnyConnection implements Connection /** @var ChannelInterface|null */ private $transactionalChannel; + /** @var float|null */ + private $lastOperationAt; + + /** @var int */ + private $runningOperations = 0; + public function __construct(Configuration\Connection $configuration) { $this->configuration = new BunnyConfiguration( @@ -42,6 +60,8 @@ public function __construct(Configuration\Connection $configuration) timeout: $configuration->getConnectionTimeout(), heartbeat: (float) $configuration->getHeartbeat(), ); + $this->operationTimeout = $configuration->getOperationTimeout(); + $this->heartbeat = $configuration->getHeartbeat(); $this->client = new Client($this->configuration); } @@ -92,9 +112,7 @@ public function connect(): void // itself as connected yet refuses to connect again - so every later attempt would await a // promise nothing can resolve. Such a client is unusable; start over with a fresh one. if ($this->client->isConnected()) { - $this->channel = null; - $this->transactionalChannel = null; - $this->client = new Client($this->configuration); + $this->discardClient(); } try { @@ -108,21 +126,45 @@ public function connect(): void public function disconnect(): void { - // Bunny only tolerates disconnect() on a fully connected client - canDisconnect() also - // rules out the connecting/disconnecting states that isConnected() reports as connected. - if (! $this->client->canDisconnect()) { - return; + // The connection.close handshake first, for what it flushes rather than for the courtesy: a + // publish() only reaches the socket once the loop turns again, and the local teardown below + // closes the stream with those bytes still in React's write buffer, which drops them. The + // broker then logs a connection that vanished without a connection.close and the message is + // simply gone - the same for an acknowledge issued as the last thing a consumer does. The + // handshake awaits a reply, so the loop turns and the buffer goes out ahead of it. + // + // Bounded by run(), because this is where a broker that has gone away must not be able to + // hold the process: it runs on kernel.terminate and console.terminate. Skipped for a + // connection idle past its heartbeat, which the broker has closed already - nothing + // buffered on it can arrive any more, and run() would only replace it and then find nothing + // to disconnect. + if ($this->client->canDisconnect() && ! $this->isStale()) { + try { + $this->run(function (): void { + $this->client->disconnect(0, 'Connection closed'); + }); + } catch (Throwable) { + // Nothing to add: the teardown below needs no broker, and run() has discarded the + // client already. + } } - $this->run(function (): void { - $this->client->disconnect(); - }); - - $this->channel = null; - $this->transactionalChannel = null; + $this->discardClient(); } /** + * {@inheritDoc} + * + * Bounded by the configured operation timeout, and given up on as soon as the client reports an + * error, so that a broker which stops answering - or a socket silently blackholed by a firewall + * - cannot park the caller forever. Bunny awaits every protocol reply on a promise that only an + * incoming frame settles: nothing rejects it when the other end goes away. + * + * Giving up leaves the operation's Fiber suspended inside the event loop for good, since there + * is nothing left to resume it. The client holding it is discarded here so those references die + * with the process instead of accumulating, but that Fiber is a real - if bounded - leak on the + * failure path. + * * @param Closure(): T $operation * * @return T @@ -131,11 +173,233 @@ public function disconnect(): void */ public function run(Closure $operation): mixed { - // Always on a Fiber of its own, even when one is already current (an acknowledge issued - // from inside a consumer callback). Running the operation on the calling Fiber instead - // saves an allocation, but makes PHP unable to switch back out of contexts that forbid it - // - a signal handler above all - turning a survivable shutdown into a fatal FiberError. - return await(async($operation)()); + $this->replaceStaleClient(); + + // Not only zero: a negative bound would be a timer already due, ending every operation + // before it starts, so anything non-positive is taken as the opt-out documented in + // docs/Setup.md. Nothing rejects a negative value, and this is why it need not. + if ($this->operationTimeout <= 0.0) { + return $this->runWithoutTimeout($operation); + } + + $client = $this->client; + $timeout = $this->operationTimeout; + + /** @var Deferred $failed */ + $failed = new Deferred(); + // What the race cannot say on its own. An exception from the caller's own closure comes out + // of the same await as a failure of ours, and Bunny re-emits every *channel* error onto the + // client (Client::channel()), so a client 'error' is not proof that the connection is gone + // either. + $failedAsynchronously = false; + $timedOut = false; + // Bunny reports asynchronous failures as an 'error' event, and Evenement drops an event + // nobody listens to - which is how a connection lost mid-operation becomes a promise that + // stays pending for the rest of the process. + $onError = static function (Throwable $error) use ($failed, &$failedAsynchronously): void { + $failedAsynchronously = true; + + $failed->reject($error); + }; + $timer = Loop::addTimer( + $timeout, + static function () use ($failed, $timeout, &$timedOut): void { + $timedOut = true; + + $failed->reject(OperationFailed::timedOut($timeout)); + }, + ); + $client->on('error', $onError); + $this->runningOperations++; + + try { + $result = await(race([async($operation)(), $failed->promise()])); + } catch (Throwable $exception) { + // A timed-out operation is still parked somewhere inside the loop, so that client is + // spent whatever its state says. Otherwise only a connection that is actually gone is + // worth replacing: Bunny tears the client down itself when the broker closes the + // connection or the socket dies, which is what leaves it unable to disconnect. + // + // Neither an exception from the caller's own closure nor a channel-level error says + // anything about the connection, and discarding on those costs a reconnect plus + // whatever was still unflushed elsewhere on it - a channel the broker closed is + // replaced on its own 'close' event. + if ($timedOut || ($failedAsynchronously && ! $client->canDisconnect())) { + $this->discardClient(); + } + + throw $exception; + } finally { + $this->runningOperations--; + Loop::cancelTimer($timer); + // The client is cached and reused, so a listener left behind would pile up one per + // operation - on the client this operation ran against, which may no longer be ours. + $client->removeListener('error', $onError); + } + + // Whether the operation turned the loop on its way is no help here - opening the channel + // does, and the publish that follows it is buffered all the same - so the turn comes after + // the operation, always. Nested calls included: the one they are nested in does turn the + // loop eventually, but a delivery callback that publishes and then works for a minute holds + // it for that minute, and a connection lost meanwhile takes the unwritten message with it. + $this->flushWrites(); + + $this->lastOperationAt = microtime(true); + + return $result; + } + + /** + * {@inheritDoc} + * + * @param Closure(): T $operation + * + * @return T + * + * @template T + */ + public function runWithoutTimeout(Closure $operation): mixed + { + $this->replaceStaleClient(); + + $result = $this->awaitOnItsOwnFiber($operation); + + // As in run(). The consume loop needs it at the very end above all: once the wait is over + // nothing turns the loop again, and the last thing written is usually an acknowledge. + $this->flushWrites(); + + return $result; + } + + /** + * Always on a Fiber of its own, even when one is already current (an acknowledge issued from + * inside a consumer callback). Running the operation on the calling Fiber instead saves an + * allocation, but makes PHP unable to switch back out of contexts that forbid it - a signal + * handler above all - turning a survivable shutdown into a fatal FiberError. + * + * @param Closure(): T $operation + * + * @return T + * + * @template T + */ + private function awaitOnItsOwnFiber(Closure $operation): mixed + { + $this->runningOperations++; + + try { + $result = await(async($operation)()); + } finally { + $this->runningOperations--; + } + + $this->lastOperationAt = microtime(true); + + return $result; + } + + /** + * Replaces a connection that has been idle for too long, unless an operation of this + * connection's own is already in flight: that one is driving the loop, heartbeats are flowing, + * and pulling its connection away mid-flight is the last thing it needs. + * + * Counted rather than read off Fiber::getCurrent(), which only stands for "nested" until the + * application brings Fibers of its own - anything built on React, or any other Fiber-based + * runtime. Such a caller is not nested at all, and would have the check skipped for good. + */ + private function replaceStaleClient(): void + { + if ($this->runningOperations > 0 || ! $this->isStale()) { + return; + } + + $this->discardClient(); + } + + /** + * Tears the client down without talking to the broker and replaces it, so that whatever comes + * next starts from a connection known to be new. + * + * RAW_CONNECTION_INACTIVE closes the channels locally and skips the connection.close handshake, + * so it cannot block on a broker that is already gone. It is also the only path that reaches + * Bunny's Connection::disconnect(), which cancels the heartbeat timer - the timer that would + * otherwise keep the event loop, and with it the whole process, alive with no stream left to + * ever wake it. + */ + private function discardClient(): void + { + if ($this->client->canDisconnect()) { + try { + // Unbounded on purpose: there is no round trip to wait for, and abandoning this + // halfway would leave the heartbeat timer armed - the very thing it is here for. + $this->awaitOnItsOwnFiber(function (): void { + $this->client->disconnect( + 0, + 'Connection discarded', + ClientInterface::RAW_CONNECTION_INACTIVE, + ); + }); + } catch (Throwable) { + // Nothing useful left to do - the client is being thrown away either way. + } + } + + $this->channel = null; + $this->transactionalChannel = null; + $this->client = new Client($this->configuration); + $this->lastOperationAt = null; + } + + /** + * Gives the event loop the one iteration it takes to put a write on the socket. + * + * publish(), ack() and nack() await no reply, so an operation made of them alone never suspends + * and the loop never turns - and React only writes its buffer when an iteration finds the socket + * writable. Bunny awaits a drain of its own once that buffer passes React's soft limit, which is + * 64 KiB, or a body of 65488 bytes with the smallest framing there is; below that a producer of + * ordinary messages sends nothing at all. Measured: twelve publishes half a second apart reached + * the broker as none of them, and with heartbeat=1 the broker hung up on a producer that had + * been publishing the whole time - while lastOperationAt below was refreshed by every one of + * them, so the connection never counted as stale either. + * + * A timer whose callback asks for a future tick, rather than either on its own: an await that a + * tick or a timer resolves resumes this Fiber from inside the loop's own tick or timer phase, + * before it reaches the stream_select() that does the writing. Asking for the tick from the + * timer leaves that queue non-empty instead, so the loop polls the streams with no timeout on + * its way to it. Measured, and the reason neither futureTick() nor delay(0) will do here. + */ + private function flushWrites(): void + { + /** @var Deferred $flushed */ + $flushed = new Deferred(); + + Loop::addTimer(0.0, static fn () => Loop::futureTick(static fn () => $flushed->resolve(null))); + + // Counted as an operation for the length of it, so that whatever the loop dispatches in that + // iteration - a delivery callback of the application's own that publishes, say - does not + // find the connection stale and pull it away from under this one. + $this->runningOperations++; + + try { + await($flushed->promise()); + } finally { + $this->runningOperations--; + } + } + + /** + * Whether the connection has been idle for longer than the heartbeat it promised the broker. + * + * Nothing drives the event loop between operations, so a connection idle for that long cannot + * have sent a single heartbeat frame and the broker has almost certainly closed it. One + * interval rather than the two the broker waits for is deliberate: reconnecting early costs a + * handshake, publishing into a dead socket costs the message. The interval is always positive - + * Configuration\Connection rejects anything else, since Bunny cannot switch heartbeats off. + */ + private function isStale(): bool + { + return $this->lastOperationAt !== null + && microtime(true) - $this->lastOperationAt >= $this->heartbeat; } private function createChannel(): ChannelInterface diff --git a/src/RabbitMQ/Connection.php b/src/RabbitMQ/Connection.php index 1ceac66..b0a75dd 100644 --- a/src/RabbitMQ/Connection.php +++ b/src/RabbitMQ/Connection.php @@ -27,6 +27,9 @@ public function disconnect(): void; * - a signal handler above all - turning a survivable shutdown into a fatal FiberError. Nesting * run() calls is therefore allowed, and each nested call gets its own Fiber. * + * The operation has to be bounded: nothing in Bunny gives up on a broker that stops answering, + * so an implementation that waits forever hands the caller a process that can never finish. + * * @param Closure(): T $operation * * @return T the value returned by $operation @@ -34,4 +37,16 @@ public function disconnect(): void; * @template T */ public function run(Closure $operation): mixed; + + /** + * Same as run(), for the one operation that is meant to block: the consume loop, which returns + * when the consumer's own message or time limit says so. Everything else wants run(). + * + * @param Closure(): T $operation + * + * @return T the value returned by $operation + * + * @template T + */ + public function runWithoutTimeout(Closure $operation): mixed; } diff --git a/src/RabbitMQ/Operation/GetOperation.php b/src/RabbitMQ/Operation/GetOperation.php index 556c6df..6a72a44 100644 --- a/src/RabbitMQ/Operation/GetOperation.php +++ b/src/RabbitMQ/Operation/GetOperation.php @@ -17,24 +17,31 @@ public function __construct(Connection $connection) $this->connection = $connection; } - /** @return Message[] */ + /** + * A bounded run() per basic.get rather than one around the lot, for the reason SetupAction + * declares one item at a time: operation_timeout is what a single round trip may take, and + * every get here is one. Sharing a budget between thousands of them ends a read the broker + * answered promptly throughout with an OperationFailed - and discards the connection with it. + * Measured against a real broker at ~0.1ms a get: 5000 of them ran out of a 0.5s bound. + * + * @return Message[] + */ public function handle(string $queueName, int $maxCount): array { - return $this->connection->run(function () use ($queueName, $maxCount): array { - $messages = []; - $channel = $this->connection->getChannel(); + $messages = []; - for ($count = 0; $count < $maxCount; $count++) { - $message = $channel->get($queueName, false); + for ($count = 0; $count < $maxCount; $count++) { + $message = $this->connection->run( + fn () => $this->connection->getChannel()->get($queueName, false), + ); - if ($message === null) { - return $messages; - } - - $messages[] = $message; + if ($message === null) { + return $messages; } - return $messages; - }); + $messages[] = $message; + } + + return $messages; } } diff --git a/src/RabbitMQ/Operation/PublishOperation.php b/src/RabbitMQ/Operation/PublishOperation.php index 298ceb6..ce3b195 100644 --- a/src/RabbitMQ/Operation/PublishOperation.php +++ b/src/RabbitMQ/Operation/PublishOperation.php @@ -4,6 +4,7 @@ namespace Cdn77\RabbitMQBundle\RabbitMQ\Operation; +use Cdn77\RabbitMQBundle\Exception\Exception as BundleException; use Cdn77\RabbitMQBundle\Exception\OperationFailed; use Cdn77\RabbitMQBundle\RabbitMQ\Connection; use Cdn77\RabbitMQBundle\RabbitMQ\Message; @@ -48,43 +49,61 @@ public function handle(Connection $connection, Message $message, string $routing }); } - /** @param Message[] $messages */ + /** + * The wrapping sits around run(), not inside the operation, because the failure does not + * always come back through the operation's own Fiber. A broker that closes the channel gets + * the rollback attempt below as far as awaiting its tx.rollback-ok, and that suspension hands + * the read loop the very channel.close that caused all this: it reaches the channel, whose + * 'error' the client re-emits, and run() then loses the race to a Fiber that is still parked in + * the rollback - which is where the wrapper used to be. Verified against a real broker: a batch + * published to a missing exchange leaked Bunny's ChannelException. + * + * @param Message[] $messages + */ public function handleAll( Connection $connection, iterable $messages, string $routingKey, string $exchangeName, ): void { - $connection->run(static function () use ($connection, $messages, $routingKey, $exchangeName): void { - $transactionalChannel = $connection->getTransactionalChannel(); - try { - foreach ($messages as $message) { - $transactionalChannel->publish( - $message->body, - $message->headers, - $exchangeName, - $routingKey, - self::MANDATORY, - self::IMMEDIATE, - ); - } - - $transactionalChannel->txCommit(); - } catch (Throwable $exception) { + try { + $connection->run(static function () use ($connection, $messages, $routingKey, $exchangeName): void { + $transactionalChannel = $connection->getTransactionalChannel(); try { - $transactionalChannel->txRollback(); - } catch (Throwable) { - // The usual cause of the failure above is the broker closing the channel, and - // such a channel can no longer be rolled back - nor does it need to be. Keep - // reporting what actually went wrong. - } + foreach ($messages as $message) { + $transactionalChannel->publish( + $message->body, + $message->headers, + $exchangeName, + $routingKey, + self::MANDATORY, + self::IMMEDIATE, + ); + } - throw new OperationFailed( - $exception->getMessage(), - $exception->getCode(), - $exception, - ); - } - }); + $transactionalChannel->txCommit(); + } catch (Throwable $exception) { + try { + $transactionalChannel->txRollback(); + } catch (Throwable) { + // The usual cause of the failure above is the broker closing the channel, + // and such a channel can no longer be rolled back - nor does it need to be. + // Keep reporting what actually went wrong. + } + + throw $exception; + } + }); + } catch (BundleException $exception) { + // Already ours, and more specific than this operation could be: a connection that could + // not be established, or the timeout that ended the commit. + throw $exception; + } catch (Throwable $exception) { + throw new OperationFailed( + $exception->getMessage(), + $exception->getCode(), + $exception, + ); + } } } diff --git a/src/SetupAction.php b/src/SetupAction.php index 760831b..c8753a2 100644 --- a/src/SetupAction.php +++ b/src/SetupAction.php @@ -19,77 +19,91 @@ public function __construct(Connection $connection) $this->connection = $connection; } + /** + * A bounded run() per declaration rather than one around the lot. operation_timeout is what a + * single round trip may take, and a topology of hundreds of items would otherwise run out of it + * while every one of them was answered promptly. + * + * It also puts the timeout where it can be reported: run() raises it from outside the + * operation's own Fiber, which stays suspended in the frame it is waiting for, so a catch + * inside the closure never sees it. Wrapped around the whole topology, a single declaration + * that hung left setup() throwing OperationFailed with nothing to say about which item it was. + */ public function setup(Topology $topology): void { - $this->connection->run(function () use ($topology): void { - $this->declareTopology($topology); - }); + $this->declareExchanges($topology); + $this->declareQueues($topology); } - private function declareTopology(Topology $topology): void + private function declareExchanges(Topology $topology): void { - $channel = $this->connection->getChannel(); - foreach ($topology->getExchanges() as $exchange) { try { - $channel->exchangeDeclare( - $exchange->getName(), - $exchange->getExchangeType()->getValue(), - false, - $exchange->isDurable(), - $exchange->shouldAutoDelete(), - $exchange->isInternal(), - false, - $exchange->getArguments(), - ); + $this->connection->run(function () use ($exchange): void { + $this->connection->getChannel()->exchangeDeclare( + $exchange->getName(), + $exchange->getExchangeType()->getValue(), + false, + $exchange->isDurable(), + $exchange->shouldAutoDelete(), + $exchange->isInternal(), + false, + $exchange->getArguments(), + ); + }); } catch (Throwable $exception) { throw ConfigurationFailed::cannotDeclareExchange($exchange, $exception); } foreach ($exchange->getBindings() as $binding) { - $boundQueue = $binding->getBindable(); - try { - $channel->exchangeBind( - $exchange->getName(), - $boundQueue->getName(), - $binding->getRoutingKey(), - false, - $binding->getArguments(), - ); + $this->connection->run(function () use ($exchange, $binding): void { + $this->connection->getChannel()->exchangeBind( + $exchange->getName(), + $binding->getBindable()->getName(), + $binding->getRoutingKey(), + false, + $binding->getArguments(), + ); + }); } catch (Throwable $exception) { throw ConfigurationFailed::cannotBindExchange($exchange, $binding, $exception); } } } + } + private function declareQueues(Topology $topology): void + { foreach ($topology->getQueues() as $queue) { try { - $channel->queueDeclare( - $queue->getName(), - false, - $queue->isDurable(), - $queue->isExclusive(), - $queue->shouldAutoDelete(), - false, - $queue->getArguments(), - ); + $this->connection->run(function () use ($queue): void { + $this->connection->getChannel()->queueDeclare( + $queue->getName(), + false, + $queue->isDurable(), + $queue->isExclusive(), + $queue->shouldAutoDelete(), + false, + $queue->getArguments(), + ); + }); } catch (Throwable $exception) { throw ConfigurationFailed::cannotDeclareQueue($queue, $exception); } foreach ($queue->getBindings() as $binding) { - $boundQueue = $binding->getBindable(); - try { - // Bunny 0.6 changed queue.bind argument order to (exchange, queue). - $channel->queueBind( - $boundQueue->getName(), - $queue->getName(), - $binding->getRoutingKey(), - false, - $binding->getArguments(), - ); + $this->connection->run(function () use ($queue, $binding): void { + // Bunny 0.6 changed queue.bind argument order to (exchange, queue). + $this->connection->getChannel()->queueBind( + $binding->getBindable()->getName(), + $queue->getName(), + $binding->getRoutingKey(), + false, + $binding->getArguments(), + ); + }); } catch (Throwable $exception) { throw ConfigurationFailed::cannotBindQueue($queue, $binding, $exception); } diff --git a/tests/Configuration/ConnectionTest.php b/tests/Configuration/ConnectionTest.php new file mode 100644 index 0000000..500facd --- /dev/null +++ b/tests/Configuration/ConnectionTest.php @@ -0,0 +1,41 @@ + self::DSN, + Configuration::KEY_CONFIGURATION_HEARTBEAT => 0, + ]); + } +} diff --git a/tests/ConsumerRunnerFailureTest.php b/tests/ConsumerRunnerFailureTest.php index 0effa91..e7b2792 100644 --- a/tests/ConsumerRunnerFailureTest.php +++ b/tests/ConsumerRunnerFailureTest.php @@ -11,12 +11,15 @@ use Bunny\Protocol\MethodBasicQosOkFrame; use Cdn77\RabbitMQBundle\ConsumerRunner; use Cdn77\RabbitMQBundle\Exception\ConfigurationFailed; +use Cdn77\RabbitMQBundle\Exception\OperationFailed; use Cdn77\RabbitMQBundle\RabbitMQ\Connection; use Cdn77\RabbitMQBundle\RabbitMQ\Consumer\Configuration; use Cdn77\RabbitMQBundle\RabbitMQ\Operation\AcknowledgeOperation; use Cdn77\RabbitMQBundle\Tests\RabbitMQ\InMemoryConsumer; use Closure; use PHPUnit\Framework\Attributes\AllowMockObjectsWithoutExpectations; +use PHPUnit\Framework\Attributes\PreserveGlobalState; +use PHPUnit\Framework\Attributes\RunTestsInSeparateProcesses; use PHPUnit\Framework\TestCase; use React\EventLoop\Loop; @@ -29,13 +32,20 @@ * 0.6.0-alpha.4 breaks React's await() scheduler for whoever took one, so an integration test * could not even clean up after itself. Partial mocks of Bunny's Channel keep its real event * emitter, which is what the ordering below hangs on. + * + * In their own processes: these emit into the shared event loop from a future tick, and an + * operation that ends in a rejection can leave that tick unconsumed - it would then fire inside + * some later test's await() and fail it with an exception from here. */ +#[RunTestsInSeparateProcesses] +#[PreserveGlobalState(false)] // A partial mock rather than a stub on purpose: stubbing every method would take Bunny's event // emitter with it, and the emitting is the point. Nothing here asserts a call. #[AllowMockObjectsWithoutExpectations] final class ConsumerRunnerFailureTest extends TestCase { private const string CHANNEL_CLOSED = 'Channel closed by server: NOT_FOUND - no queue \'aQueue\''; + private const float OPERATION_TIMEOUT = 0.2; public function testQosFailureIsTranslated(): void { @@ -44,8 +54,11 @@ public function testQosFailureIsTranslated(): void $connection = $this->givenConnectionTo($channel); - // A non-zero prefetch-size, the rejection RabbitMQ is guaranteed to answer with. Nothing - // is ever consumed, so the acknowledge operation is only there to build the consumer. + // A non-zero prefetch-size, the rejection RabbitMQ is guaranteed to answer with - though + // only as far as the values go: a channel that throws where the real one is resumed with + // the refusal by Bunny's await list. That path is covered live by + // ConsumerRunnerTest::testPrefetchRefusedByTheBrokerIsTranslated. Nothing is ever consumed, + // so the acknowledge operation is only there to build the consumer. $consumer = new InMemoryConsumer( new AcknowledgeOperation($connection), new Configuration('aQueue', 1, 1), @@ -84,22 +97,63 @@ static function () use ($channel): MethodBasicConsumeOkFrame { new Configuration('aQueue'), ); - // Not ConnectionFailed::channelClosed(), which says nothing about why. + // Not ConnectionFailed::channelClosedWithoutAnError(), which says nothing about this one. self::expectException(ChannelException::class); self::expectExceptionMessage(self::CHANNEL_CLOSED); (new ConsumerRunner($connection))->run($consumer); } + /** + * Opening the channel, basic.qos and basic.consume each wait for the broker to answer, and + * Bunny leaves such a wait pending for good when the socket dies - the handlers the runner + * installs are no help, a Fiber stuck in the handshake never reaches the await() they reject. + * So the startup goes through the bounded run(), and whatever that reports ends the run. + */ + public function testStartupIsBoundedByTheOperationTimeout(): void + { + $channel = $this->createPartialMock(Channel::class, ['qos', 'consume', 'cancel']); + $channel->method('qos')->willReturn(new MethodBasicQosOkFrame()); + $channel->method('cancel')->willReturn(false); + $channel->method('consume')->willReturnCallback( + static function (): MethodBasicConsumeOkFrame { + $consumeOk = new MethodBasicConsumeOkFrame(); + $consumeOk->consumerTag = 'aConsumerTag'; + + return $consumeOk; + }, + ); + + $connection = self::createStub(Connection::class); + $connection->method('run')->willThrowException(OperationFailed::timedOut(self::OPERATION_TIMEOUT)); + $connection->method('runWithoutTimeout')->willReturnCallback( + static fn (Closure $operation) => await(async($operation)()), + ); + $connection->method('getChannel')->willReturn($channel); + + // A time limit shorter than the suite can afford to wait: a startup that went unbounded + // would find a broker that answers everything and nothing to consume, and had to end + // somewhere. + $consumer = new InMemoryConsumer( + new AcknowledgeOperation($connection), + new Configuration('aQueue', 1, 0, null, self::OPERATION_TIMEOUT), + ); + + self::expectException(OperationFailed::class); + self::expectExceptionMessage('did not finish within'); + + (new ConsumerRunner($connection))->run($consumer); + } + private function givenConnectionTo(Channel $channel): Connection { $connection = self::createStub(Connection::class); // Like BunnyConnection: a Fiber of its own, so that the runner's own await() suspends that // Fiber rather than the main context - the one place React's scheduler Fiber is shared with // whatever else ran before in this process. - $connection->method('run')->willReturnCallback( - static fn (Closure $operation) => await(async($operation)()), - ); + $onRun = static fn (Closure $operation) => await(async($operation)()); + $connection->method('run')->willReturnCallback($onRun); + $connection->method('runWithoutTimeout')->willReturnCallback($onRun); $connection->method('getChannel')->willReturn($channel); return $connection; diff --git a/tests/ConsumerRunnerTest.php b/tests/ConsumerRunnerTest.php index 7b99718..a31b02f 100644 --- a/tests/ConsumerRunnerTest.php +++ b/tests/ConsumerRunnerTest.php @@ -5,8 +5,13 @@ namespace Cdn77\RabbitMQBundle\Tests; use Bunny\Message; +use Cdn77\RabbitMQBundle\Configuration\Connection; +use Cdn77\RabbitMQBundle\Configuration\Dsn; use Cdn77\RabbitMQBundle\Configuration\Topology; +use Cdn77\RabbitMQBundle\ConsumerRunner; +use Cdn77\RabbitMQBundle\Exception\ConfigurationFailed; use Cdn77\RabbitMQBundle\RabbitMQ\Binding; +use Cdn77\RabbitMQBundle\RabbitMQ\BunnyConnection; use Cdn77\RabbitMQBundle\RabbitMQ\Consumer\Configuration; use Cdn77\RabbitMQBundle\RabbitMQ\Consumer\Consumer; use Cdn77\RabbitMQBundle\RabbitMQ\Exchange; @@ -14,19 +19,42 @@ use Cdn77\RabbitMQBundle\RabbitMQ\Operation\AcknowledgeOperation; use Cdn77\RabbitMQBundle\RabbitMQ\Queue; use Cdn77\RabbitMQBundle\Tests\RabbitMQ\InMemoryConsumer; +use Cdn77\RabbitMQBundle\Tests\RabbitMQ\SuspendingConsumer; use Cdn77\RabbitMQBundle\Tests\RabbitMQ\ThrowingConsumer; use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\Attributes\Group; use PHPUnit\Framework\TestCase; use RuntimeException; +use function assert; use function end; +use function fgets; +use function getenv; +use function is_string; +use function microtime; +use function proc_close; +use function proc_open; +use function proc_terminate; +use function stream_get_contents; +use function usleep; + +use const PHP_BINARY; #[Group('Integration')] final class ConsumerRunnerTest extends TestCase { use WithRabbitMQ; + private const float SHORT_OPERATION_TIMEOUT = 0.2; + private const float CONSUMING_SECONDS = 1.0; + private const float STOP_AFTER_SECONDS = 0.3; + private const float SUSPEND_FOR_SECONDS = 1.0; + private const string HANDLER_SOURCE_QUEUE = 'consumerRunnerTestSourceQueue'; + private const string HANDLER_TARGET_QUEUE = 'consumerRunnerTestTargetQueue'; + private const int SIGKILL = 9; + private const float BROKER_SECONDS = 3.0; + private const int POLL_MICROSECONDS = 50000; + /** @return int[][] */ public static function maxMessagesDataProvider(): array { @@ -96,6 +124,215 @@ public function testConsumerIsNotCalledAgainAfterItFailed(): void self::assertSame(1, $consumer->getConsumeCallCount()); } + /** + * The consume loop is the one operation that must not be bounded: it returns when the + * consumer's own message or time limit says so, which is well past any per-operation timeout. + * Bounding it would end every consumer with an OperationFailed instead. + */ + public function testConsumingIsNotBoundedByTheOperationTimeout(): void + { + $exchange = new Exchange('test', new ExchangeType(ExchangeType::DIRECT)); + $queue = $this->givenEmptyQueue($exchange, 'a_routing_key'); + + // An operation timeout far shorter than the time the consumer is told to wait for messages + // that are never going to come. + $connection = new BunnyConnection(Connection::fromDsn( + new Dsn(self::dsn() . '&operation_timeout=' . self::SHORT_OPERATION_TIMEOUT), + )); + $consumer = new InMemoryConsumer( + new AcknowledgeOperation($connection), + new Configuration($queue->getName(), 1, 0, null, self::CONSUMING_SECONDS), + ); + + $startedAt = microtime(true); + (new ConsumerRunner($connection))->run($consumer); + $consumingTook = microtime(true) - $startedAt; + + $connection->disconnect(); + + self::assertGreaterThanOrEqual(self::CONSUMING_SECONDS, $consumingTook); + self::assertCount(0, $consumer->getConsumedMessages()); + } + + /** + * The prefetch failure as a real broker reports it, which the unit test's throwing channel + * cannot stand in for: RabbitMQ answers a non-zero prefetch-size with a connection.close, and + * that frame is dispatched twice - Bunny's await list rejects the pending basic.qos-ok with it + * first (resuming this operation's Fiber from inside the read loop, so the translation below + * happens), and only then does it reach the channel, which emits 'close' and an 'error' the + * client re-emits. Should that order ever turn around, the bounded run() around the startup + * would win the race and leak Bunny's exception instead. + * + * On a connection of its own: this one is closed by the broker, and the test class shares + * another with its own teardown. + */ + public function testPrefetchRefusedByTheBrokerIsTranslated(): void + { + $queue = $this->givenEmptyQueue( + new Exchange('test', new ExchangeType(ExchangeType::DIRECT)), + 'a_routing_key', + ); + + $connection = new BunnyConnection(Connection::fromDsn(new Dsn(self::dsn()))); + $consumer = new InMemoryConsumer( + new AcknowledgeOperation($connection), + new Configuration($queue->getName(), 1, 1), + ); + + // Only the catch around qos() raises this one, so getting it is the whole point: had the + // race been won by the client error the same refusal ends up as Bunny's ClientException. + self::expectException(ConfigurationFailed::class); + self::expectExceptionMessage('Could not set prefetch-size/prefetch-count'); + + (new ConsumerRunner($connection))->run($consumer); + } + + /** + * The time limit falling due while a message is being handled must not end the run there. A + * handler that awaits anything suspends its own Fiber and turns the loop, which is where the + * timer fires: ending the run then had run() return, the consumer cancelled and the connection + * closed - by kernel.terminate or console.terminate - with the handler still parked, so the + * acknowledge it went on to issue was lost and the message redelivered. + */ + public function testStopWaitsForAMessageStillBeingHandled(): void + { + // One message only, so that what is left on the queue afterwards can only be this one. + $queue = $this->givenQueueWithOneMessage(); + $consumer = new SuspendingConsumer( + new AcknowledgeOperation($this->getConnection()), + // A limit that falls due while the handler below still holds the message. + new Configuration($queue->getName(), 1, 0, null, self::STOP_AFTER_SECONDS), + self::SUSPEND_FOR_SECONDS, + ); + + $startedAt = microtime(true); + $this->whenConsume($consumer); + $consumingTook = microtime(true) - $startedAt; + + self::assertGreaterThanOrEqual(self::SUSPEND_FOR_SECONDS, $consumingTook); + // The acknowledge reached the broker, so the message is gone rather than back on the queue. + self::assertSame(0, $this->whenCountingMessagesLeft($queue)); + } + + /** + * And the same handler failing still fails the run. Settling the promise when the timer fired + * left this rejection landing on one that had settled long ago, where react/promise drops it - + * the command exited successfully having half-processed a message. + */ + public function testExceptionFromAMessageStillBeingHandledIsPropagated(): void + { + $queue = $this->givenQueueWithOneMessage(); + $consumer = new SuspendingConsumer( + new AcknowledgeOperation($this->getConnection()), + new Configuration($queue->getName(), 1, 0, null, self::STOP_AFTER_SECONDS), + self::SUSPEND_FOR_SECONDS, + true, + ); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage(SuspendingConsumer::EXCEPTION_MESSAGE); + + $this->whenConsume($consumer); + } + + /** + * A handler that publishes and then works on for a while - synchronous PHP, so it turns the + * loop no more than the acknowledge after it does - had its message sit in React's write buffer + * until it returned, and lose it altogether if the connection went away first. + */ + public function testPublishFromAHandlerIsOnTheSocketBeforeTheHandlerReturns(): void + { + $this->givenEmptyQueues(); + + $process = proc_open( + [PHP_BINARY, __DIR__ . '/fixtures/publish-from-a-blocking-handler.php'], + [1 => ['pipe', 'w'], 2 => ['pipe', 'w']], + $pipes, + null, + [ + 'RABBITMQ_DSN' => self::dsn(), + 'SOURCE_QUEUE_NAME' => self::HANDLER_SOURCE_QUEUE, + 'TARGET_QUEUE_NAME' => self::HANDLER_TARGET_QUEUE, + ], + ); + self::assertIsResource($process); + + $published = (string) fgets($pipes[1]); + // Asked for while the handler is still blocked, so that only a flush from inside it can + // answer - and before the kill, which a process cannot survive to flush anything either. + $count = $this->whenCountingMessagesOn(self::HANDLER_TARGET_QUEUE); + + proc_terminate($process, self::SIGKILL); + $errors = (string) stream_get_contents($pipes[2]); + proc_close($process); + $this->thenQueuesAreGone(); + + self::assertStringContainsString('published', $published, $errors); + self::assertSame(1, $count); + } + + private static function dsn(): string + { + $dsn = getenv('RABBITMQ_DSN'); + + assert(is_string($dsn)); + + return $dsn; + } + + private function whenCountingMessagesLeft(Queue $queue): int + { + $connection = $this->getConnection(); + + // Declaring a queue that exists is how AMQP asks how many messages are ready on it. + return $connection->run( + static fn (): int => $connection->getChannel()->queueDeclare($queue->getName())->messageCount, + ); + } + + private function givenEmptyQueues(): void + { + $connection = $this->getConnection(); + $connection->run(static function () use ($connection): void { + foreach ([self::HANDLER_SOURCE_QUEUE, self::HANDLER_TARGET_QUEUE] as $queueName) { + $connection->getChannel()->queueDeclare($queueName); + $connection->getChannel()->queuePurge($queueName); + } + }); + } + + private function thenQueuesAreGone(): void + { + $connection = $this->getConnection(); + $connection->run(static function () use ($connection): void { + foreach ([self::HANDLER_SOURCE_QUEUE, self::HANDLER_TARGET_QUEUE] as $queueName) { + $connection->getChannel()->queueDelete($queueName); + } + }); + } + + private function whenCountingMessagesOn(string $queueName): int + { + $connection = $this->getConnection(); + $deadline = microtime(true) + self::BROKER_SECONDS; + + // More than once, because the broker counts a message when it has got round to it, not when + // its bytes landed on the socket a moment earlier. + do { + $count = $connection->run( + static fn (): int => $connection->getChannel()->queueDeclare($queueName)->messageCount, + ); + + if ($count > 0) { + break; + } + + usleep(self::POLL_MICROSECONDS); + } while (microtime(true) < $deadline); + + return $count; + } + private function clearRabbitMQ(): void { $connection = $this->getConnection(); @@ -108,8 +345,31 @@ private function clearRabbitMQ(): void private function givenQueueWithEnoughMessages(): Queue { $exchange = new Exchange('test', new ExchangeType(ExchangeType::DIRECT)); - $queue = new Queue('testQueue'); $routingKey = 'a_routing_key'; + $queue = $this->givenEmptyQueue($exchange, $routingKey); + + $this->givenEnoughMessagesInQueue($exchange, $routingKey); + + return $queue; + } + + private function givenQueueWithOneMessage(): Queue + { + $exchange = new Exchange('test', new ExchangeType(ExchangeType::DIRECT)); + $routingKey = 'a_routing_key'; + $queue = $this->givenEmptyQueue($exchange, $routingKey); + + $connection = $this->getConnection(); + $connection->run(static function () use ($connection, $exchange, $routingKey): void { + $connection->getChannel()->publish('a message', [], $exchange->getName(), $routingKey); + }); + + return $queue; + } + + private function givenEmptyQueue(Exchange $exchange, string $routingKey): Queue + { + $queue = new Queue('testQueue'); $topology = new Topology( [$exchange], [], @@ -118,8 +378,6 @@ private function givenQueueWithEnoughMessages(): Queue ); $this->setupTopology($topology); - $this->givenEnoughMessagesInQueue($exchange, $routingKey); - return $queue; } diff --git a/tests/EventListener/DisconnectConnectionTest.php b/tests/EventListener/DisconnectConnectionTest.php index c0b62ce..2d2e524 100644 --- a/tests/EventListener/DisconnectConnectionTest.php +++ b/tests/EventListener/DisconnectConnectionTest.php @@ -8,7 +8,7 @@ use Cdn77\RabbitMQBundle\RabbitMQ\Connection; use Fiber; use PHPUnit\Framework\Attributes\PreserveGlobalState; -use PHPUnit\Framework\Attributes\RunInSeparateProcess; +use PHPUnit\Framework\Attributes\RunTestsInSeparateProcesses; use PHPUnit\Framework\TestCase; use stdClass; use Symfony\Component\Console\Command\Command; @@ -19,10 +19,36 @@ use Symfony\Component\EventDispatcher\EventDispatcher; use Symfony\Component\HttpKernel\KernelEvents; +use function dirname; +use function fclose; +use function microtime; +use function proc_close; +use function proc_get_status; +use function proc_open; +use function proc_terminate; +use function stream_get_contents; +use function stream_socket_get_name; +use function stream_socket_server; +use function usleep; + +use const PHP_BINARY; + +/** + * Every test here runs in its own process: terminating stops the event loop, which is process-wide + * and outlives the test - React's shared scheduler Fiber would be left in a run() that returns the + * moment anything resumes it, and the next await() in the suite would die with + * `AssertionError: assert(\is_callable($ret))`. + */ +#[RunTestsInSeparateProcesses] +#[PreserveGlobalState(false)] final class DisconnectConnectionTest extends TestCase { /** SIGTERM, spelled out so that the test does not need the pcntl extension. */ private const int SIGNAL = 15; + private const float HANDSHAKE_TIMEOUT = 1.0; + private const float EXIT_ALLOWANCE = 5.0; + private const int POLL_MICROSECONDS = 20000; + private const int SIGKILL = 9; public function testDisconnectsOnHttpAndConsoleTermination(): void { @@ -37,14 +63,6 @@ public function testDisconnectsOnHttpAndConsoleTermination(): void $dispatcher->dispatch($this->consoleTermination(null), ConsoleEvents::TERMINATE); } - /** - * In its own process: the subscriber stops the event loop, which is process-wide and outlives - * the test - React's shared scheduler Fiber is left in a run() that returns the moment anything - * resumes it, so the next await() in the suite would die with - * `AssertionError: assert(\is_callable($ret))`. - */ - #[RunInSeparateProcess] - #[PreserveGlobalState(false)] public function testDoesNotDisconnectWhenTerminatingFromSignalHandler(): void { $connection = $this->createMock(Connection::class); @@ -74,6 +92,62 @@ public function testDisconnectsFromInsideFiber(): void self::assertTrue($fiber->isTerminated()); } + /** + * The Loop::stop() on a normal console termination is not the belt-and-braces it looks like: a + * handshake that stalls leaves Bunny's socket on the event loop for good, since + * Client::connect() neither closes the connection nor rolls the state back when it throws + * (bunny 0.6.0-alpha.4), and disconnect() cannot take it off - the client never reached a state + * it can be disconnected from. React's shutdown then blocks in stream_select() with nothing to + * wake it and this subprocess runs until it is killed, which is what happens if that one line + * is ever removed. + * + * A subprocess because there is no other way to tell a process that exits from one that hangs, + * and because stopping the loop is process-wide. + */ + public function testCommandExitsAfterAHandshakeThatStalled(): void + { + $listener = stream_socket_server('tcp://127.0.0.1:0', $errorNumber, $errorMessage); + self::assertIsResource($listener, (string) $errorMessage); + + // Never accepted, so the connection completes from the backlog and the handshake then waits + // for a greeting that is never coming - the operation timeout is what ends it. + $address = stream_socket_get_name($listener, false); + self::assertIsString($address); + + $process = proc_open( + [PHP_BINARY, dirname(__DIR__) . '/fixtures/console-termination.php'], + [1 => ['pipe', 'w'], 2 => ['pipe', 'w']], + $pipes, + null, + [ + 'RABBITMQ_DSN' => 'amqp://' . $address + . '/?heartbeat=60&operation_timeout=' . self::HANDSHAKE_TIMEOUT, + ], + ); + self::assertIsResource($process); + + $deadline = microtime(true) + self::HANDSHAKE_TIMEOUT + self::EXIT_ALLOWANCE; + while (proc_get_status($process)['running'] === true && microtime(true) < $deadline) { + usleep(self::POLL_MICROSECONDS); + } + + $running = proc_get_status($process)['running'] === true; + + // Kill it before reading: stream_get_contents() waits for the end of a pipe that a process + // still holding it open is never going to give. + if ($running) { + proc_terminate($process, self::SIGKILL); + } + + $output = (string) stream_get_contents($pipes[1]) . (string) stream_get_contents($pipes[2]); + + proc_close($process); + fclose($listener); + + self::assertStringContainsString('connect failed', $output); + self::assertFalse($running, 'The command was still running: ' . $output); + } + private function givenDispatcher(Connection $connection): EventDispatcher { $dispatcher = new EventDispatcher(); diff --git a/tests/RabbitMQ/BunnyConnectionTest.php b/tests/RabbitMQ/BunnyConnectionTest.php index 5da8f5b..55cf5f3 100644 --- a/tests/RabbitMQ/BunnyConnectionTest.php +++ b/tests/RabbitMQ/BunnyConnectionTest.php @@ -4,21 +4,40 @@ namespace Cdn77\RabbitMQBundle\Tests\RabbitMQ; +use Bunny\Exception\ChannelException; use Cdn77\RabbitMQBundle\Configuration\Connection as ConnectionConfiguration; use Cdn77\RabbitMQBundle\Configuration\Dsn; +use Cdn77\RabbitMQBundle\Exception\OperationFailed; use Cdn77\RabbitMQBundle\RabbitMQ\BunnyConnection; +use Cdn77\RabbitMQBundle\RabbitMQ\Message; +use Cdn77\RabbitMQBundle\RabbitMQ\Operation\GetOperation; +use Cdn77\RabbitMQBundle\RabbitMQ\Operation\PublishOperation; use Cdn77\RabbitMQBundle\Tests\WithRabbitMQ; use Fiber; +use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\Attributes\Group; use PHPUnit\Framework\TestCase; +use React\Promise\Promise; +use RuntimeException; +use function getenv; +use function React\Async\async; +use function React\Async\await; use function React\Async\delay; +use function sleep; final class BunnyConnectionTest extends TestCase { use WithRabbitMQ; private const string QUEUE = 'freshChannelQueue'; + private const string STALE_QUEUE = 'staleConnectionQueue'; + private const string UNRELATED_FIBER_QUEUE = 'unrelatedFiberQueue'; + private const string UNFLUSHED_QUEUE = 'unflushedPublishQueue'; + private const string THROWN_QUEUE = 'operationThrewQueue'; + + /** Short enough to keep the suite quick, long enough for the broker to accept it. */ + private const int SHORT_HEARTBEAT = 1; /** Generous: the close frame is one round trip away, and a slow runner should not flake. */ private const float CLOSE_ARRIVAL = 1.0; @@ -35,24 +54,194 @@ public function testChannelClosedByTheBrokerIsReplaced(): void $stale = $connection->run(static fn () => $connection->getChannel()); - // RabbitMQ answers a publish to an exchange that does not exist by closing the channel. + $this->whenTheBrokerClosesTheChannel($connection); + + // A working channel, not just a different object: the broker answers on it. + $declareOk = $connection->run( + static fn () => $connection->getChannel()->queueDeclare(self::QUEUE), + ); + $connection->run(static fn () => $connection->getChannel()->queueDelete(self::QUEUE)); + + self::assertNotSame($stale, $connection->run(static fn () => $connection->getChannel())); + self::assertSame(self::QUEUE, $declareOk->queue); + } + + /** + * The channel.close carries the reason, and Bunny hands it over as an 'error' event that + * nobody used to listen to - leaving the operation it interrupted to await a reply that was + * never coming. + */ + #[Group('Integration')] + public function testChannelClosedByTheBrokerSurfacesAsAnError(): void + { + $connection = $this->getConnection(); + $connection->run( static fn () => $connection->getChannel()->publish('body', [], 'noSuchExchange', 'aKey'), ); - // That channel.close arrives on its own, with nothing awaiting it, so give the loop a turn. + self::expectException(ChannelException::class); + self::expectExceptionMessage("NOT_FOUND - no exchange 'noSuchExchange'"); + $connection->run(static fn () => delay(self::CLOSE_ARRIVAL)); + } - $fresh = $connection->run(static fn () => $connection->getChannel()); + /** @return array */ + public static function operationTimeoutProvider(): array + { + return [ + 'a fifth of a second' => [0.2, 'Operation did not finish within 0.2 seconds'], + // A bound of a fixed number of decimals would report this one as 0.0 seconds, which is + // the value that turns the bound off. + 'a twenty-fifth of a second' => [0.04, 'Operation did not finish within 0.04 seconds'], + ]; + } - // A working channel, not just a different object: the broker answers on it. - $declareOk = $connection->run( - static fn () => $connection->getChannel()->queueDeclare(self::QUEUE), + /** + * Nothing in Bunny gives up on a broker that stops answering: every protocol reply is awaited + * on a promise that only an incoming frame settles. Without a bound of our own, an operation + * against a dead-but-open socket parks the process for as long as it runs - and the bound it + * gives up at is the one it reports. + */ + #[DataProvider('operationTimeoutProvider')] + public function testOperationThatNeverFinishesTimesOut(float $operationTimeout, string $reported): void + { + $connection = new BunnyConnection($this->givenConfiguration(operationTimeout: $operationTimeout)); + + self::expectException(OperationFailed::class); + self::expectExceptionMessage($reported); + + // A promise nobody settles, which is what a lost connection looks like from in here. + $connection->run(static fn () => await(new Promise(static function (): void { + }))); + } + + /** + * The event loop only turns while an operation is awaiting, so a process that publishes and + * then goes off to do minutes of work of its own cannot send a single heartbeat in between - + * and the broker hangs up on it. Both messages have to arrive regardless. + */ + #[Group('Integration')] + public function testConnectionIdleForLongerThanTheHeartbeatIsReplaced(): void + { + $connection = new BunnyConnection($this->givenConfiguration(heartbeat: self::SHORT_HEARTBEAT)); + + $publish = new PublishOperation(); + + $connection->run(static fn () => $connection->getChannel()->queueDeclare(self::STALE_QUEUE)); + // Transactional, so that a returned call means the broker has the message - no waiting for + // an asynchronous publish to show up on the queue. + $publish->handleAll($connection, [Message::json('{"n":1}')], self::STALE_QUEUE, ''); + + // Blocking on purpose: this is the frozen loop, not a wait for anything. + sleep(self::SHORT_HEARTBEAT * 3); + + $publish->handleAll($connection, [Message::json('{"n":2}')], self::STALE_QUEUE, ''); + + $messages = (new GetOperation($connection))->handle(self::STALE_QUEUE, 2); + + $connection->run(static fn () => $connection->getChannel()->queueDelete(self::STALE_QUEUE)); + $connection->disconnect(); + + self::assertCount(2, $messages); + } + + /** + * The same, for a caller that is on a Fiber already - one of the application's own, with no + * operation of this connection anywhere below it. Whether a call is nested has to be counted: + * inferred from Fiber::getCurrent() it holds only until something brings Fibers of its own, + * and anything built on React does, at which point the check above never runs again. + */ + #[Group('Integration')] + public function testStaleConnectionIsReplacedForACallerAlreadyOnAFiber(): void + { + $connection = new BunnyConnection($this->givenConfiguration(heartbeat: self::SHORT_HEARTBEAT)); + + $publish = new PublishOperation(); + $publishSecond = static fn () => $publish->handleAll( + $connection, + [Message::json('{"n":2}')], + self::UNRELATED_FIBER_QUEUE, + '', ); - $connection->run(static fn () => $connection->getChannel()->queueDelete(self::QUEUE)); - self::assertNotSame($stale, $fresh); - self::assertSame(self::QUEUE, $declareOk->queue); + $connection->run( + static fn () => $connection->getChannel()->queueDeclare(self::UNRELATED_FIBER_QUEUE), + ); + $publish->handleAll($connection, [Message::json('{"n":1}')], self::UNRELATED_FIBER_QUEUE, ''); + + sleep(self::SHORT_HEARTBEAT * 3); + + await(async($publishSecond)()); + + $messages = (new GetOperation($connection))->handle(self::UNRELATED_FIBER_QUEUE, 2); + + $connection->run( + static fn () => $connection->getChannel()->queueDelete(self::UNRELATED_FIBER_QUEUE), + ); + $connection->disconnect(); + + self::assertCount(2, $messages); + } + + /** + * A published message survives the teardown, which is what a producer cares about: closing the + * socket locally drops whatever React still holds, and this used to lose the message and leave + * "client unexpectedly closed TCP connection" in the broker log. + * + * Two things stand behind it now and this does not tell them apart: run() turns the loop after + * every operation, so the publish is on the socket before disconnect() is even called, and + * disconnect() still asks for the connection.close handshake first, which awaits a reply and + * would flush anything left. The handshake is no longer what saves the message - it is what + * tells the broker we meant to go. + */ + #[Group('Integration')] + public function testDisconnectFlushesAPublishThatHasNotReachedTheSocketYet(): void + { + $connection = new BunnyConnection($this->givenConfiguration()); + $reader = $this->getConnection(); + + $connection->run(static fn () => $connection->getChannel()->queueDeclare(self::UNFLUSHED_QUEUE)); + // handle(), not handleAll(): a transaction commits, and that round trip would flush the + // publish on its own. + (new PublishOperation())->handle($connection, Message::json('{"n":1}'), self::UNFLUSHED_QUEUE, ''); + $connection->disconnect(); + + // Read on a connection of its own - the broker's view of what actually arrived. + $messages = (new GetOperation($reader))->handle(self::UNFLUSHED_QUEUE, 1); + + $reader->run(static fn () => $reader->getChannel()->queueDelete(self::UNFLUSHED_QUEUE)); + + self::assertCount(1, $messages); + } + + /** + * An exception from the caller's own closure leaves the await by the same door as a connection + * failure, and used to be taken for one: the client was discarded, which costs a reconnect and + * - the teardown being local - whatever was still in the write buffer. The connection is not + * what failed here, so it stays, and so does the publish from before. + */ + #[Group('Integration')] + public function testAnOperationThatThrowsOfItsOwnAccordKeepsTheConnection(): void + { + $connection = new BunnyConnection($this->givenConfiguration()); + $reader = $this->getConnection(); + + $connection->run(static fn () => $connection->getChannel()->queueDeclare(self::THROWN_QUEUE)); + $channel = $connection->run(static fn () => $connection->getChannel()); + (new PublishOperation())->handle($connection, Message::json('{"n":1}'), self::THROWN_QUEUE, ''); + + $this->whenTheOperationThrows($connection); + + $channelAfter = $connection->run(static fn () => $connection->getChannel()); + // Whatever is left in the buffer goes out here, as it would at the end of a request. + $connection->disconnect(); + + $messages = (new GetOperation($reader))->handle(self::THROWN_QUEUE, 1); + $reader->run(static fn () => $reader->getChannel()->queueDelete(self::THROWN_QUEUE)); + + self::assertSame($channel, $channelAfter); + self::assertCount(1, $messages); } /** @@ -89,4 +278,53 @@ protected function tearDown(): void { $this->getConnection()->disconnect(); } + + /** The broker this suite runs against, with the timings a single test needs. */ + private function givenConfiguration( + int $heartbeat = 60, + float $operationTimeout = 30.0, + ): ConnectionConfiguration { + $dsn = new Dsn((string) getenv('RABBITMQ_DSN')); + + return new ConnectionConfiguration( + $dsn->getHost(), + $dsn->getPort(), + $dsn->getVhost(), + $dsn->getUsername(), + $dsn->getPassword(), + $heartbeat, + operationTimeout: $operationTimeout, + ); + } + + /** + * RabbitMQ answers a publish to an exchange that does not exist by closing the channel. The + * failure surfaces on whatever operation comes next - here, only a wait for it to arrive - + * which is what testChannelClosedByTheBrokerSurfacesAsAnError() is about. + */ + + /** The failure itself is nothing to do with the broker, and is asserted on in its own right. */ + private function whenTheOperationThrows(BunnyConnection $connection): void + { + try { + $connection->run(static function (): void { + throw new RuntimeException('a failure of the caller\'s own'); + }); + } catch (RuntimeException) { + // Expected: it is the connection afterwards that this is about. + } + } + + private function whenTheBrokerClosesTheChannel(BunnyConnection $connection): void + { + $connection->run( + static fn () => $connection->getChannel()->publish('body', [], 'noSuchExchange', 'aKey'), + ); + + try { + $connection->run(static fn () => delay(self::CLOSE_ARRIVAL)); + } catch (ChannelException) { + // Expected, and asserted on its own elsewhere. + } + } } diff --git a/tests/RabbitMQ/Operation/GetOperationTest.php b/tests/RabbitMQ/Operation/GetOperationTest.php new file mode 100644 index 0000000..b4a7673 --- /dev/null +++ b/tests/RabbitMQ/Operation/GetOperationTest.php @@ -0,0 +1,84 @@ +givenQueueWithMessages(); + + $reader = new BunnyConnection(Connection::fromDsn( + new Dsn(self::dsn() . '&operation_timeout=' . self::OPERATION_TIMEOUT), + )); + + $startedAt = microtime(true); + $messages = (new GetOperation($reader))->handle(self::QUEUE, self::MESSAGE_COUNT); + $readTook = microtime(true) - $startedAt; + + $reader->disconnect(); + + self::assertCount(self::MESSAGE_COUNT, $messages); + self::assertGreaterThan(self::OPERATION_TIMEOUT, $readTook); + } + + protected function tearDown(): void + { + $connection = new BunnyConnection(Connection::fromDsn(new Dsn(self::dsn()))); + $connection->run(static function () use ($connection): void { + $connection->getChannel()->queueDelete(self::QUEUE); + }); + $connection->disconnect(); + + parent::tearDown(); + } + + private function givenQueueWithMessages(): void + { + $connection = new BunnyConnection(Connection::fromDsn(new Dsn(self::dsn()))); + $connection->run(static function () use ($connection): void { + $channel = $connection->getChannel(); + $channel->queueDeclare(self::QUEUE); + $channel->queuePurge(self::QUEUE); + + for ($count = 0; $count < self::MESSAGE_COUNT; $count++) { + $channel->publish((string) $count, [], '', self::QUEUE); + } + }); + $connection->disconnect(); + } + + private static function dsn(): string + { + $dsn = getenv('RABBITMQ_DSN'); + + assert(is_string($dsn)); + + return $dsn; + } +} diff --git a/tests/RabbitMQ/Operation/PublishOperationTest.php b/tests/RabbitMQ/Operation/PublishOperationTest.php new file mode 100644 index 0000000..64b2bf7 --- /dev/null +++ b/tests/RabbitMQ/Operation/PublishOperationTest.php @@ -0,0 +1,149 @@ +handleAll( + $connection, + [Message::json('{"a":1}'), Message::json('{"a":2}')], + 'a_routing_key', + 'no-such-exchange', + ); + } + + /** + * A publish has to be on the socket by the time handle() returns, which takes a turn of the + * event loop: publish() awaits no reply, so nothing suspends and React holds the bytes in its + * write buffer until an iteration finds the socket writable. Bunny only awaits a drain of its + * own once that buffer passes 64 KiB - a body of 65488 bytes with the smallest framing - so + * without the turn a producer of ordinary messages sends nothing at all, and the broker + * eventually hangs up on it for missing heartbeats it never had the chance to write. + * + * A subprocess that is killed rather than allowed to exit: nothing in this process may turn the + * loop before the broker is asked, and an orderly exit would run React's shutdown and flush the + * buffer whether the publish had managed it or not. + */ + public function testPublishIsOnTheSocketWhenItReturns(): void + { + $this->givenEmptyQueue(); + + $process = proc_open( + [PHP_BINARY, dirname(__DIR__, 2) . '/fixtures/publish-without-flushing.php'], + [1 => ['pipe', 'w'], 2 => ['pipe', 'w']], + $pipes, + null, + ['RABBITMQ_DSN' => self::dsn(), 'QUEUE_NAME' => self::UNFLUSHED_QUEUE], + ); + self::assertIsResource($process); + + $published = (string) fgets($pipes[1]); + proc_terminate($process, self::SIGKILL); + $errors = (string) stream_get_contents($pipes[2]); + proc_close($process); + + self::assertStringContainsString('published', $published, $errors); + $this->thenTheMessageReachedTheBroker(); + } + + private function givenEmptyQueue(): void + { + $connection = new BunnyConnection(Connection::fromDsn(new Dsn(self::dsn()))); + $connection->run(static function () use ($connection): void { + $channel = $connection->getChannel(); + $channel->queueDeclare(self::UNFLUSHED_QUEUE); + $channel->queuePurge(self::UNFLUSHED_QUEUE); + }); + $connection->disconnect(); + } + + private function thenTheMessageReachedTheBroker(): void + { + $connection = new BunnyConnection(Connection::fromDsn(new Dsn(self::dsn()))); + $deadline = microtime(true) + self::BROKER_SECONDS; + + // Declaring a queue that exists is how AMQP asks how many messages are ready on it - and + // asked more than once, because the broker counts a message when it has got round to it, + // not when its bytes landed on the socket a moment earlier. + do { + $count = $connection->run( + static fn (): int => $connection->getChannel()->queueDeclare(self::UNFLUSHED_QUEUE)->messageCount, + ); + + if ($count > 0) { + break; + } + + usleep(self::POLL_MICROSECONDS); + } while (microtime(true) < $deadline); + + $connection->run(static fn () => $connection->getChannel()->queueDelete(self::UNFLUSHED_QUEUE)); + $connection->disconnect(); + + self::assertSame(1, $count); + } + + private static function dsn(): string + { + $dsn = getenv('RABBITMQ_DSN'); + + assert(is_string($dsn)); + + return $dsn; + } +} diff --git a/tests/RabbitMQ/SuspendingConsumer.php b/tests/RabbitMQ/SuspendingConsumer.php new file mode 100644 index 0000000..59e770e --- /dev/null +++ b/tests/RabbitMQ/SuspendingConsumer.php @@ -0,0 +1,68 @@ +acknowledgeOperation = $acknowledgeOperation; + $this->configuration = $configuration; + $this->suspendForSeconds = $suspendForSeconds; + $this->throwWhenResumed = $throwWhenResumed; + } + + public function consume(Message $message): void + { + delay($this->suspendForSeconds); + + if ($this->throwWhenResumed) { + throw new RuntimeException(self::EXCEPTION_MESSAGE); + } + + $this->acknowledgeOperation->handle($message); + } + + public function getName(): string + { + return 'suspending'; + } + + public function getConfiguration(): Configuration + { + return $this->configuration; + } +} diff --git a/tests/SetupActionTest.php b/tests/SetupActionTest.php new file mode 100644 index 0000000..152f70b --- /dev/null +++ b/tests/SetupActionTest.php @@ -0,0 +1,61 @@ +givenConnectionTimingOut(); + $topology = new Topology( + [new Exchange(self::EXCHANGE, new ExchangeType(ExchangeType::DIRECT))], + [], + [], + [], + ); + + self::expectException(ConfigurationFailed::class); + self::expectExceptionMessage('Could not declare exchange ' . self::EXCHANGE); + + (new SetupAction($connection))->setup($topology); + } + + public function testTimedOutQueueDeclarationNamesTheQueue(): void + { + $connection = $this->givenConnectionTimingOut(); + $topology = new Topology([], [], [new Queue(self::QUEUE)], []); + + self::expectException(ConfigurationFailed::class); + self::expectExceptionMessage('Could not declare queue ' . self::QUEUE); + + (new SetupAction($connection))->setup($topology); + } + + private function givenConnectionTimingOut(): Connection + { + $connection = self::createStub(Connection::class); + $connection->method('run')->willThrowException(OperationFailed::timedOut(30.0)); + + return $connection; + } +} diff --git a/tests/fixtures/console-termination.php b/tests/fixtures/console-termination.php new file mode 100644 index 0000000..7c62c3d --- /dev/null +++ b/tests/fixtures/console-termination.php @@ -0,0 +1,43 @@ +addSubscriber(new DisconnectConnection($connection)); + +try { + $connection->connect(); +} catch (Throwable $error) { + echo 'connect failed: ', $error::class, "\n"; +} + +$dispatcher->dispatch( + new ConsoleTerminateEvent(new Command('test'), new ArrayInput([]), new NullOutput(), 0, null), + ConsoleEvents::TERMINATE, +); + +echo "terminated\n"; diff --git a/tests/fixtures/publish-from-a-blocking-handler.php b/tests/fixtures/publish-from-a-blocking-handler.php new file mode 100644 index 0000000..d5c8bed --- /dev/null +++ b/tests/fixtures/publish-from-a-blocking-handler.php @@ -0,0 +1,64 @@ +handleRaw($connection, 'a message to consume', [], $sourceQueueName, ''); + +$consumer = new class ($connection, $publishOperation, $sourceQueueName, $targetQueueName) implements Consumer { + public function __construct( + private BunnyConnection $connection, + private PublishOperation $publishOperation, + private string $sourceQueueName, + private string $targetQueueName, + ) { + } + + public function consume(Message $message): void + { + $this->publishOperation->handleRaw($this->connection, 'a message', [], $this->targetQueueName, ''); + + echo "published\n"; + + // Killed from the outside while sitting here, so that the handler returning can never be + // what put the message above on the socket. + sleep(30); + } + + public function getName(): string + { + return 'blockingHandler'; + } + + public function getConfiguration(): Configuration + { + return new Configuration($this->sourceQueueName); + } +}; + +(new ConsumerRunner($connection))->run($consumer); diff --git a/tests/fixtures/publish-without-flushing.php b/tests/fixtures/publish-without-flushing.php new file mode 100644 index 0000000..aba1ff4 --- /dev/null +++ b/tests/fixtures/publish-without-flushing.php @@ -0,0 +1,32 @@ +handleRaw($connection, 'a message', [], $queueName, ''); + +echo "published\n"; + +// Killed from the outside while sitting here: exiting would run React's shutdown, which turns the +// loop and would flush the write buffer whether the publish had managed it or not. Long enough that +// the test is always the one to end this, short enough to leave nothing behind if it is not. +sleep(30);