Reproducible bug with amqp basic_cancel

composer.json

@@ -33,6 +33,7 @@
     "require-dev": {
         "infection/infection": "^0.27.8||^0.29.0",
         "maglnet/composer-require-checker": "^4.4",
+        "phpbench/phpbench": "^1.4",
         "phpunit/phpunit": "^9.5",
         "rector/rector": "^2.0.15",
         "roave/infection-static-analysis-plugin": "^1.16",

phpbench.json

@@ -0,0 +1,8 @@
+{
+    "$schema":"./vendor/phpbench/phpbench/phpbench.schema.json",
+    "runner.bootstrap": "vendor/autoload.php",
+    "runner.path": "tests/Benchmark",
+    "runner.revs": 100000,
+    "runner.iterations": 5,
+    "runner.warmup": 5
+}

src/ExistingMessagesConsumer.php

@@ -29,34 +29,36 @@
      */
     public function consume(callable $callback): void
     {
-        $this->channel->basic_consume(
-            $this->queueName,
-            '',
-            false,
-            false,
-            false,
-            false,
-            function (AMQPMessage $amqpMessage) use ($callback): void {
-                try {
-                    $message = $this->serializer->unserialize($amqpMessage->getBody());
-                    if ($this->messageConsumed = $callback($message)) {
-                        $this->channel->basic_ack($amqpMessage->getDeliveryTag());
-                    }
-                } catch (Throwable $exception) {
-                    $this->messageConsumed = false;
-                    $consumerTag = $amqpMessage->getConsumerTag();
-                    if ($consumerTag !== null) {
-                        $this->channel->basic_cancel($consumerTag);
-                    }
+        $consumerTag = uniqid(more_entropy: true);
+        try {
+            $this->channel->basic_consume(
+                $this->queueName,
+                $consumerTag,
+                false,
+                false,
+                false,
+                false,
+                function (AMQPMessage $amqpMessage) use ($callback): void {
+                    try {
+                        $message = $this->serializer->unserialize($amqpMessage->getBody());
+                        if ($this->messageConsumed = $callback($message)) {
+                            $this->channel->basic_ack($amqpMessage->getDeliveryTag());
+                        }
+                    } catch (Throwable $exception) {
+                        $this->messageConsumed = false;
+                        $this->channel->basic_nack($amqpMessage->getDeliveryTag(), false, true);
 
-                    throw $exception;
+                        throw $exception;
+                    }
                 }
-            }
-        );
+            );
 
-        do {
-            $this->messageConsumed = false;
-            $this->channel->wait(null, true);
-        } while ($this->messageConsumed === true);
+            do {
+                $this->messageConsumed = false;
+                $this->channel->wait(null, true);
+            } while ($this->messageConsumed === true);
+        } finally {
+            $this->channel->basic_cancel($consumerTag, false, false);
+        }
     }
 }

tests/Benchmark/QueueBench.php

@@ -0,0 +1,72 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Benchmark;
+
+use PhpAmqpLib\Connection\AMQPStreamConnection;
+use PhpBench\Attributes\OutputMode;
+use PhpBench\Attributes\OutputTimeUnit;
+use PhpBench\Attributes\Skip;
+use Yiisoft\Queue\AMQP\Adapter;
+use Yiisoft\Queue\AMQP\QueueProvider;
+use Yiisoft\Queue\AMQP\Settings\Queue as QueueSettings;
+use Yiisoft\Queue\Cli\SimpleLoop;
+use Yiisoft\Queue\Message\JsonMessageSerializer;
+use Yiisoft\Queue\Message\Message;
+
+final class QueueBench
+{
+    private const CONSUME_REVISIONS = 100_000;
+
+    private Adapter $adapter;
+
+    public function __construct()
+    {
+        $loop = new SimpleLoop();
+        $serializer = new JsonMessageSerializer();
+        $queueProvider = new QueueProvider(
+            new AMQPStreamConnection(
+                getenv('RABBITMQ_HOST'),
+                getenv('RABBITMQ_PORT'),
+                getenv('RABBITMQ_USER'),
+                getenv('RABBITMQ_PASSWORD'),
+            ),
+            new QueueSettings()
+        );
+        $adapter = new Adapter($queueProvider, $serializer, $loop);
+
+        $this->adapter = $adapter;
+    }
+
+    #[\PhpBench\Attributes\BeforeMethods('cleanupQueue')]
+    #[OutputMode('throughput')]
+    #[OutputTimeUnit('seconds', 3)]
+    #[Skip]
+    public function benchPush(): void
+    {
+        $this->adapter->push(new Message('test', ['payload' => 'test']));
+    }
+
+    public function cleanupQueue(): void
+    {
+        $this->adapter->runExisting(static fn (): bool => true);
+    }
+
+    #[\PhpBench\Attributes\Iterations(5)]
+    #[\PhpBench\Attributes\Revs(self::CONSUME_REVISIONS)]
+    #[\PhpBench\Attributes\BeforeMethods('pushMessagesForConsume')]
+    #[OutputMode('throughput')]
+    #[OutputTimeUnit('seconds', 3)]
+    public function benchConsume(): void
+    {
+        $this->adapter->runExisting(static fn (): bool => false);
+    }
+
+    public function pushMessagesForConsume(): void
+    {
+        for ($i = 0; $i < self::CONSUME_REVISIONS; $i++) {
+            $this->adapter->push(new Message('test', ['payload' => 'test']));
+        }
+    }
+}