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/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/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..1a12dc1 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,19 @@ 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. + +> **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 33bdd62..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&read_write_timeout=3` +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/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..f67ca6b 100644 --- a/src/Configuration/Connection.php +++ b/src/Configuration/Connection.php @@ -5,12 +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 int DEFAULT_READ_WRITE_TIMEOUT = 5; + private const float DEFAULT_OPERATION_TIMEOUT = 30.0; /** @var string */ private $host; @@ -33,8 +36,8 @@ final class Connection /** @var int */ private $connectionTimeout; - /** @var int */ - private $readWriteTimeout; + /** @var float */ + private $operationTimeout; public function __construct( string $host, @@ -44,16 +47,16 @@ public function __construct( string|null $password, int $heartbeat = self::DEFAULT_HEARTBEAT, int $connectionTimeout = self::DEFAULT_CONNECTION_TIMEOUT, - int $readWriteTimeout = self::DEFAULT_READ_WRITE_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->readWriteTimeout = $readWriteTimeout; + $this->operationTimeout = $operationTimeout; } /** @param mixed[] $configuration */ @@ -66,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 ( @@ -76,11 +81,14 @@ 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 ( - isset($configuration[Configuration::KEY_CONFIGURATION_READ_WRITE_TIMEOUT]) - && ! isset($dsn->getParameters()[Configuration::KEY_CONFIGURATION_READ_WRITE_TIMEOUT]) + is_numeric($operationTimeout) + && ! isset($dsn->getParameters()[Configuration::KEY_CONFIGURATION_OPERATION_TIMEOUT]) ) { - $new->readWriteTimeout = (int) $configuration[Configuration::KEY_CONFIGURATION_READ_WRITE_TIMEOUT]; + $new->operationTimeout = (float) $operationTimeout; } return $new; @@ -89,6 +97,7 @@ public static function fromDI(array $configuration): self public static function fromDsn(Dsn $dsn): self { $parameters = $dsn->getParameters(); + $operationTimeout = $parameters[Configuration::KEY_CONFIGURATION_OPERATION_TIMEOUT] ?? null; return new self( $dsn->getHost(), @@ -99,8 +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), - (int) ($parameters[Configuration::KEY_CONFIGURATION_READ_WRITE_TIMEOUT] - ?? self::DEFAULT_READ_WRITE_TIMEOUT), + is_numeric($operationTimeout) ? (float) $operationTimeout : self::DEFAULT_OPERATION_TIMEOUT, ); } @@ -139,8 +147,27 @@ public function getConnectionTimeout(): int return $this->connectionTimeout; } - public function getReadWriteTimeout(): int + /** How long a single broker operation may take before it is given up on. Zero disables it. */ + public function getOperationTimeout(): float { - return $this->readWriteTimeout; + 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 fa65313..e1f86e1 100644 --- a/src/ConsumerRunner.php +++ b/src/ConsumerRunner.php @@ -4,16 +4,18 @@ namespace Cdn77\RabbitMQBundle; -use Bunny\Channel; -use Bunny\Client; +use Bunny\ChannelInterface; 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 +32,218 @@ 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); + // 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. + // + // 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; - $this->processedMessageCount++; - - if ($this->hasAnyMessageLeft($consumerConfig->getMaxMessages(), $this->processedMessageCount)) { return; } - $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); - } + 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()), + ); + + // 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, + $stop, + $fail, + $closed, + $stopped, + &$stopping, + &$handling, + &$stopWhenHandled, + ): array { + $channel = $this->connection->getChannel(); + + try { + $channel->qos($configuration->getPrefetchSize(), $configuration->getPrefetchCount()); + } catch (Throwable $error) { + throw ConfigurationFailed::invalidPrefetchValues($error); + } + + $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]; + }, + ); + + $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); + } - 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($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..ea0d293 100644 --- a/src/DependencyInjection/Configuration.php +++ b/src/DependencyInjection/Configuration.php @@ -16,7 +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_READ_WRITE_TIMEOUT = 'read_write_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'; @@ -35,7 +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 DEAFULT_READ_WRITE_TIMEOUT = 3; + private const int DEFAULT_OPERATION_TIMEOUT = 30; public function getConfigTreeBuilder(): TreeBuilder { @@ -65,8 +65,8 @@ private function configureConnection(ArrayNodeDefinition $rootNode): void ->defaultValue(self::DEFAULT_TIMEOUT); $rootNode->children() - ->scalarNode(self::KEY_CONFIGURATION_READ_WRITE_TIMEOUT) - ->defaultValue(self::DEAFULT_READ_WRITE_TIMEOUT); + ->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 new file mode 100644 index 0000000..63118d7 --- /dev/null +++ b/src/EventListener/DisconnectConnection.php @@ -0,0 +1,89 @@ +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(); + + 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(); + } + + /** + * 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..3785076 100644 --- a/src/Exception/ConfigurationFailed.php +++ b/src/Exception/ConfigurationFailed.php @@ -8,28 +8,37 @@ 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 heartbeatMustBePositive(int $heartbeat): self { - return new self('Could not set prefetch-size/prefetch-count'); + return new self(sprintf('Heartbeat must be a positive number of seconds, %d given', $heartbeat)); } - public static function cannotDeclareExchange(Exchange $exchange): self + public static function invalidPrefetchValues(Throwable|null $previous = null): self { - return new self(sprintf('Could not declare exchange %s', $exchange->getName())); + return new self('Could not set prefetch-size/prefetch-count', 0, $previous); } - public static function cannotDeclareQueue(Queue $queue): self + public static function cannotDeclareExchange(Exchange $exchange, Throwable|null $previous = null): self { - return new self(sprintf('Could not declare queue %s', $queue->getName())); + return new self(sprintf('Could not declare exchange %s', $exchange->getName()), 0, $previous); } - public static function cannotBindExchange(Exchange $exchange, Binding $binding): self + public static function cannotDeclareQueue(Queue $queue, Throwable|null $previous = null): self { + return new self(sprintf('Could not declare queue %s', $queue->getName()), 0, $previous); + } + + 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 +46,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 +60,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..8564ff0 100644 --- a/src/Exception/ConnectionFailed.php +++ b/src/Exception/ConnectionFailed.php @@ -13,4 +13,18 @@ public static function causedBy(Throwable $previous): self { return new self('Connection to RabbitMQ failed', 0, $previous); } + + /** + * 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 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 989afcc..7e7bc59 100644 --- a/src/Exception/OperationFailed.php +++ b/src/Exception/OperationFailed.php @@ -5,13 +5,20 @@ namespace Cdn77\RabbitMQBundle\Exception; use RuntimeException; +use Throwable; use function sprintf; final class OperationFailed extends RuntimeException implements Exception { - public static function gotInvalidType(string $expected, string $actual): self + public static function timedOut(float $seconds, Throwable|null $previous = null): self { - return new self(sprintf('Expected "%s", got "%s"', $expected, $actual)); + 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/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..616e01a 100644 --- a/src/RabbitMQ/BunnyConnection.php +++ b/src/RabbitMQ/BunnyConnection.php @@ -4,66 +4,99 @@ namespace Cdn77\RabbitMQBundle\RabbitMQ; -use Bunny\Channel; +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 React\Promise\PromiseInterface; +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; - /** @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(); - } + /** @var float|null */ + private $lastOperationAt; - if ($configuration->getPassword() !== null) { - $options['password'] = $configuration->getPassword(); - } + /** @var int */ + private $runningOperations = 0; - $this->client = new Client($options); + public function __construct(Configuration\Connection $configuration) + { + $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->operationTimeout = $configuration->getOperationTimeout(); + $this->heartbeat = $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 +104,21 @@ 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->discardClient(); + } + try { - $this->client->connect(); + $this->run(function (): void { + $this->client->connect(); + }); } catch (Throwable $exception) { throw ConnectionFailed::causedBy($exception); } @@ -84,24 +126,286 @@ public function connect(): void public function disconnect(): void { - if (! $this->client->isConnected()) { + // 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->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 + * + * @template T + */ + public function run(Closure $operation): mixed + { + $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->client->disconnect(); + $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; } - private function createChannel(): Channel + /** + * 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 { - $this->connect(); + /** @var Deferred $flushed */ + $flushed = new Deferred(); + + Loop::addTimer(0.0, static fn () => Loop::futureTick(static fn () => $flushed->resolve(null))); - $channel = $this->client->channel(); + // 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++; - if ($channel instanceof PromiseInterface) { - throw CannotCreateChannel::gotInvalidType(Channel::class, PromiseInterface::class); + try { + await($flushed->promise()); + } finally { + $this->runningOperations--; } + } - return $channel; + /** + * 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 + { + $this->connect(); + + return $this->client->channel(); } } diff --git a/src/RabbitMQ/Connection.php b/src/RabbitMQ/Connection.php index b99798a..b0a75dd 100644 --- a/src/RabbitMQ/Connection.php +++ b/src/RabbitMQ/Connection.php @@ -4,15 +4,49 @@ 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. + * + * 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 + * + * @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/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..6a72a44 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 { @@ -19,23 +17,28 @@ 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 { $messages = []; for ($count = 0; $count < $maxCount; $count++) { - /** @var Message|PromiseInterface|null $message */ - $message = $this->connection->getChannel()->get($queueName, false); + $message = $this->connection->run( + fn () => $this->connection->getChannel()->get($queueName, false), + ); if ($message === null) { return $messages; } - if ($message instanceof PromiseInterface) { - throw OperationFailed::gotInvalidType(Message::class, PromiseInterface::class); - } - $messages[] = $message; } diff --git a/src/RabbitMQ/Operation/PublishOperation.php b/src/RabbitMQ/Operation/PublishOperation.php index 3403fa0..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; @@ -14,7 +15,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,52 +23,82 @@ 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 */ + /** + * 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 { - $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 $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(), 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..c8753a2 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 { @@ -22,70 +19,93 @@ 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 { - $channel = $this->connection->getChannel(); + $this->declareExchanges($topology); + $this->declareQueues($topology); + } + private function declareExchanges(Topology $topology): void + { 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 { + $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(); - - $frame = $channel->exchangeBind( - $exchange->getName(), - $boundQueue->getName(), - $binding->getRoutingKey(), - false, - $binding->getArguments(), - ); - - if (! ($frame instanceof MethodExchangeBindOkFrame)) { - throw ConfigurationFailed::cannotBindExchange($exchange, $binding); + try { + $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) { - $frame = $channel->queueDeclare( - $queue->getName(), - false, - $queue->isDurable(), - $queue->isExclusive(), - $queue->shouldAutoDelete(), - false, - $queue->getArguments(), - ); - - if (! ($frame instanceof MethodQueueDeclareOkFrame)) { - throw ConfigurationFailed::cannotDeclareQueue($queue); + try { + $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(); - $frame = $channel->queueBind( - $queue->getName(), - $boundQueue->getName(), - $binding->getRoutingKey(), - false, - $binding->getArguments(), - ); - - if (! ($frame instanceof MethodQueueBindOkFrame)) { - throw ConfigurationFailed::cannotBindQueue($queue, $binding); + try { + $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 new file mode 100644 index 0000000..e7b2792 --- /dev/null +++ b/tests/ConsumerRunnerFailureTest.php @@ -0,0 +1,161 @@ +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 - 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), + ); + + 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::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. + $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 84181fe..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,17 +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 { @@ -41,44 +71,325 @@ 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()); + } + + /** + * 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 = new Queue('testQueue'); - $routingKey = 'a_routing_key'; - $topology = new Topology( - [$exchange], - [], - [$queue], - [$queue->getName() => [new Binding($exchange, $routingKey)]], + $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), ); - $this->setupTopology($topology); - $this->givenEnoughMessagesInQueue($exchange, $routingKey); - $consumer = $this->givenConfiguredConsumer($maxMessages, $queue); + $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; - $this->thenOnlyMaxMessagesCountIsConsumed($maxMessages, $consumer); + 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 { - $this->getConnection()->getChannel()->queueDelete('testQueue'); - $this->getConnection()->getChannel()->exchangeDelete('test'); + $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)); + $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], + [], + [$queue], + [$queue->getName() => [new Binding($exchange, $routingKey)]], + ); + $this->setupTopology($topology); + + 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..2d2e524 --- /dev/null +++ b/tests/EventListener/DisconnectConnectionTest.php @@ -0,0 +1,169 @@ +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); + } + + 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()); + } + + /** + * 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(); + $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..55cf5f3 --- /dev/null +++ b/tests/RabbitMQ/BunnyConnectionTest.php @@ -0,0 +1,330 @@ +getConnection(); + + $stale = $connection->run(static fn () => $connection->getChannel()); + + $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'), + ); + + self::expectException(ChannelException::class); + self::expectExceptionMessage("NOT_FOUND - no exchange 'noSuchExchange'"); + + $connection->run(static fn () => delay(self::CLOSE_ARRIVAL)); + } + + /** @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'], + ]; + } + + /** + * 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()->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); + } + + /** + * 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(); + } + + /** 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/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/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/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; + } +} 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);