[Bug] Artisan command keep running even if the consumer has exception with Broken Pipe

[sujit-baniya]

• Laravel/Lumen version: latest
• RabbitMQ version: latest
• Package version: latest

Describe the bug
Whenever there's error on Connection and throw exception php artisan rabbitmq:consume doesn't stop or terminate. As the command is running, the monitoring tools like Supervisor doesn't know the status of consumer and consumers are not restarted.
Meanwhile this exception the there's no consumer on queue but Supervisor shows the process is running

Steps To Reproduce

Start the command php artisan rabbitmq:consume and externally trigger Broken Pipe or connection

Expected behavior

Whenever there's broken pipe exception, the artisan command should have been stopped or terminated

[bandgeekndb]

This behavior also is happening with Laravel Horizon. If something happens to the connection between the queue workers and the Rabbit server, my logs start filling up with:

Broken pipe or closed connection {"exception":"[object] (PhpAmqpLib\\Exception\\AMQPConnectionClosedException(code: 0): Broken pipe or closed connection at /var/www/servicebridge/vendor/php-amqp lib/php-amqplib/PhpAmqpLib/Wire/IO/StreamIO.php:172)

and

Channel connection is closed. {"exception":"[object] (PhpAmqpLib\\Exception\\AMQPChannelClosedException(code: 0): Channel connection is closed. at /var/www/servicebridge/vendor/php-amqplib/php-a mqplib/PhpAmqpLib/Channel/AbstractChannel.php:245)

messages.

Terminating Horizon and letting supervisor or systemd restart the processes clears the errors (assuming connectivity has been reestablished between the queue workers and RabbitMQ).

[bandgeekndb]

Full stack trace:

[2021-08-19 00:44:58] production.ERROR: Broken pipe or closed connection {"exception":"[object] (PhpAmqpLib\\Exception\\AMQPConnectionClosedException(code: 0): Broken pipe or closed connection at /var/www/servicebridge/vendor/php-amqp
lib/php-amqplib/PhpAmqpLib/Wire/IO/StreamIO.php:172)
[stacktrace]
#0 /var/www/servicebridge/vendor/php-amqplib/php-amqplib/PhpAmqpLib/Wire/AMQPReader.php(157): PhpAmqpLib\\Wire\\IO\\StreamIO->read()
#1 /var/www/servicebridge/vendor/php-amqplib/php-amqplib/PhpAmqpLib/Wire/AMQPReader.php(103): PhpAmqpLib\\Wire\\AMQPReader->rawread()
#2 /var/www/servicebridge/vendor/php-amqplib/php-amqplib/PhpAmqpLib/Connection/AbstractConnection.php(580): PhpAmqpLib\\Wire\\AMQPReader->read()
#3 /var/www/servicebridge/vendor/php-amqplib/php-amqplib/PhpAmqpLib/Connection/AbstractConnection.php(636): PhpAmqpLib\\Connection\\AbstractConnection->wait_frame()
#4 /var/www/servicebridge/vendor/php-amqplib/php-amqplib/PhpAmqpLib/Channel/AbstractChannel.php(235): PhpAmqpLib\\Connection\\AbstractConnection->wait_channel()
#5 /var/www/servicebridge/vendor/php-amqplib/php-amqplib/PhpAmqpLib/Channel/AbstractChannel.php(352): PhpAmqpLib\\Channel\\AbstractChannel->next_frame()
#6 /var/www/servicebridge/vendor/php-amqplib/php-amqplib/PhpAmqpLib/Channel/AMQPChannel.php(288): PhpAmqpLib\\Channel\\AbstractChannel->wait()
#7 /var/www/servicebridge/vendor/php-amqplib/php-amqplib/PhpAmqpLib/Channel/AMQPChannel.php(118): PhpAmqpLib\\Channel\\AMQPChannel->x_open()
#8 /var/www/servicebridge/vendor/php-amqplib/php-amqplib/PhpAmqpLib/Connection/AbstractConnection.php(700): PhpAmqpLib\\Channel\\AMQPChannel->__construct()
#9 /var/www/servicebridge/vendor/vladimir-yuldashev/laravel-queue-rabbitmq/src/Queue/RabbitMQQueue.php(399): PhpAmqpLib\\Connection\\AbstractConnection->channel()
#10 /var/www/servicebridge/vendor/vladimir-yuldashev/laravel-queue-rabbitmq/src/Queue/RabbitMQQueue.php(106): VladimirYuldashev\\LaravelQueueRabbitMQ\\Queue\\RabbitMQQueue->isQueueExists()
#11 /var/www/servicebridge/vendor/vladimir-yuldashev/laravel-queue-rabbitmq/src/Horizon/RabbitMQQueue.php(34): VladimirYuldashev\\LaravelQueueRabbitMQ\\Queue\\RabbitMQQueue->size()
#12 /var/www/servicebridge/vendor/laravel/horizon/src/AutoScaler.php(80): VladimirYuldashev\\LaravelQueueRabbitMQ\\Horizon\\RabbitMQQueue->readyNow()
#13 /var/www/servicebridge/vendor/laravel/framework/src/Illuminate/Collections/Collection.php(689): Laravel\\Horizon\\AutoScaler->Laravel\\Horizon\\{closure}()
#14 /var/www/servicebridge/vendor/laravel/horizon/src/AutoScaler.php(86): Illuminate\\Support\\Collection->mapWithKeys()
#15 /var/www/servicebridge/vendor/laravel/horizon/src/AutoScaler.php(49): Laravel\\Horizon\\AutoScaler->timeToClearPerQueue()
#16 /var/www/servicebridge/vendor/laravel/horizon/src/Supervisor.php(348): Laravel\\Horizon\\AutoScaler->scale()
#17 /var/www/servicebridge/vendor/laravel/horizon/src/Supervisor.php(295): Laravel\\Horizon\\Supervisor->autoScale()
#18 /var/www/servicebridge/vendor/laravel/horizon/src/Supervisor.php(260): Laravel\\Horizon\\Supervisor->loop()
#19 /var/www/servicebridge/vendor/laravel/horizon/src/Console/SupervisorCommand.php(100): Laravel\\Horizon\\Supervisor->monitor()
#20 /var/www/servicebridge/vendor/laravel/horizon/src/Console/SupervisorCommand.php(75): Laravel\\Horizon\\Console\\SupervisorCommand->start()
#21 /var/www/servicebridge/vendor/laravel/framework/src/Illuminate/Container/BoundMethod.php(36): Laravel\\Horizon\\Console\\SupervisorCommand->handle()
#22 /var/www/servicebridge/vendor/laravel/framework/src/Illuminate/Container/Util.php(40): Illuminate\\Container\\BoundMethod::Illuminate\\Container\\{closure}()
#23 /var/www/servicebridge/vendor/laravel/framework/src/Illuminate/Container/BoundMethod.php(93): Illuminate\\Container\\Util::unwrapIfClosure()
#24 /var/www/servicebridge/vendor/laravel/framework/src/Illuminate/Container/BoundMethod.php(37): Illuminate\\Container\\BoundMethod::callBoundMethod()
#25 /var/www/servicebridge/vendor/laravel/framework/src/Illuminate/Container/Container.php(651): Illuminate\\Container\\BoundMethod::call()
#26 /var/www/servicebridge/vendor/laravel/framework/src/Illuminate/Console/Command.php(136): Illuminate\\Container\\Container->call()
#27 /var/www/servicebridge/vendor/symfony/console/Command/Command.php(299): Illuminate\\Console\\Command->execute()
#28 /var/www/servicebridge/vendor/laravel/framework/src/Illuminate/Console/Command.php(121): Symfony\\Component\\Console\\Command\\Command->run()
#29 /var/www/servicebridge/vendor/symfony/console/Application.php(978): Illuminate\\Console\\Command->run()
#30 /var/www/servicebridge/vendor/symfony/console/Application.php(295): Symfony\\Component\\Console\\Application->doRunCommand()
#31 /var/www/servicebridge/vendor/symfony/console/Application.php(167): Symfony\\Component\\Console\\Application->doRun()
#32 /var/www/servicebridge/vendor/laravel/framework/src/Illuminate/Console/Application.php(92): Symfony\\Component\\Console\\Application->run()
#33 /var/www/servicebridge/vendor/laravel/framework/src/Illuminate/Foundation/Console/Kernel.php(129): Illuminate\\Console\\Application->run()
#34 /var/www/servicebridge/artisan(37): Illuminate\\Foundation\\Console\\Kernel->handle()
#35 {main}
"}

The actual errors are happening in the php-amqplib library, is this a bug that should be raised there or do we need to add better error handling in this package for the Exceptions thrown by php-amqplib?

[yurganov]

Hello, @bandgeekndb

Did you manage to solve it?

[acosta-edgar]

We are observing the same problem, which also appears related to report #436 .
This post explains how to catch possible connection exceptions, I believe that is what needs to be done in RabbitMQConnector::connect() or RabbitMQConnector::createConnection()

[bandgeekndb]

@acosta-edgar Thanks for sharing this info, that's definitely helpful!

I'm not around next week, but when I get back, I'll fork this and see if I can add some connection recovery handling like you mentioned. If I can, I'll open a PR and see what the dev thinks!

[vyuldashev]

Should be fixed by #457