ACP2E-1747: Auto Coupon incorrectly marked as used for failed card payment

Author: bubasuma

app/code/Magento/MessageQueue/Model/CheckIsAvailableMessagesInQueue.php

@@ -7,6 +7,7 @@
 
 namespace Magento\MessageQueue\Model;
 
+use Magento\Framework\MessageQueue\CountableQueueInterface;
 use Magento\Framework\MessageQueue\QueueRepository;
 
 /**
@@ -40,6 +41,9 @@ public function __construct(QueueRepository $queueRepository)
     public function execute($connectionName, $queueName)
     {
         $queue = $this->queueRepository->get($connectionName, $queueName);
+        if ($queue instanceof CountableQueueInterface) {
+            return $queue->count() > 0;
+        }
         $message = $queue->dequeue();
         if ($message) {
             $queue->reject($message);

app/code/Magento/MessageQueue/Test/Unit/Model/CheckIsAvailableMessagesInQueueTest.php

@@ -0,0 +1,116 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+declare(strict_types=1);
+
+namespace Magento\MessageQueue\Test\Unit\Model;
+
+use Magento\Framework\MessageQueue\CountableQueueInterface;
+use Magento\Framework\MessageQueue\EnvelopeInterface;
+use Magento\Framework\MessageQueue\QueueInterface;
+use Magento\Framework\MessageQueue\QueueRepository;
+use Magento\MessageQueue\Model\CheckIsAvailableMessagesInQueue;
+use PHPUnit\Framework\MockObject\MockObject;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * Test for CheckIsAvailableMessagesInQueue
+ */
+class CheckIsAvailableMessagesInQueueTest extends TestCase
+{
+    /**
+     * @var QueueRepository|MockObject
+     */
+    private $queueRepository;
+
+    /**
+     * @var CheckIsAvailableMessagesInQueue
+     */
+    private $model;
+
+    /**
+     * @inheritdoc
+     */
+    protected function setUp(): void
+    {
+        parent::setUp();
+        $this->queueRepository = $this->createMock(QueueRepository::class);
+        $this->model = new CheckIsAvailableMessagesInQueue(
+            $this->queueRepository
+        );
+    }
+
+    public function testExecuteNotCountableAndNotEmptyQueue(): void
+    {
+        $connectionName = 'test';
+        $queueName = 'test';
+
+        $queue = $this->getMockForAbstractClass(QueueInterface::class);
+        $message = $this->getMockForAbstractClass(EnvelopeInterface::class);
+        $this->queueRepository->expects($this->once())
+            ->method('get')
+            ->with($connectionName, $queueName)
+            ->willReturn($queue);
+        $queue->expects($this->once())
+            ->method('dequeue')
+            ->willReturn($message);
+        $queue->expects($this->once())
+            ->method('reject')
+            ->willReturn($message);
+        $this->assertTrue($this->model->execute($connectionName, $queueName));
+    }
+
+    public function testExecuteNotCountableAndEmptyQueue(): void
+    {
+        $connectionName = 'test';
+        $queueName = 'test';
+
+        $queue = $this->getMockForAbstractClass(QueueInterface::class);
+        $this->queueRepository->expects($this->once())
+            ->method('get')
+            ->with($connectionName, $queueName)
+            ->willReturn($queue);
+        $queue->expects($this->once())
+            ->method('dequeue')
+            ->willReturn(null);
+        $this->assertFalse($this->model->execute($connectionName, $queueName));
+    }
+
+    public function testExecuteCountableAndNotEmptyQueue(): void
+    {
+        $connectionName = 'test';
+        $queueName = 'test';
+
+        $queue = $this->getMockForAbstractClass(CountableQueueInterface::class);
+        $this->queueRepository->expects($this->once())
+            ->method('get')
+            ->with($connectionName, $queueName)
+            ->willReturn($queue);
+        $queue->expects($this->once())
+            ->method('count')
+            ->willReturn(1);
+        $queue->expects($this->never())
+            ->method('dequeue');
+        $this->assertTrue($this->model->execute($connectionName, $queueName));
+    }
+
+    public function testExecuteCountableAndEmptyQueue(): void
+    {
+        $connectionName = 'test';
+        $queueName = 'test';
+
+        $queue = $this->getMockForAbstractClass(CountableQueueInterface::class);
+        $this->queueRepository->expects($this->once())
+            ->method('get')
+            ->with($connectionName, $queueName)
+            ->willReturn($queue);
+        $queue->expects($this->once())
+            ->method('count')
+            ->willReturn(0);
+        $queue->expects($this->never())
+            ->method('dequeue');
+        $this->assertFalse($this->model->execute($connectionName, $queueName));
+    }
+}

app/code/Magento/MysqlMq/Model/Driver/Queue.php

@@ -5,16 +5,18 @@
  */
 namespace Magento\MysqlMq\Model\Driver;
 
+use Magento\Framework\App\ObjectManager;
+use Magento\Framework\MessageQueue\CountableQueueInterface;
 use Magento\Framework\MessageQueue\EnvelopeInterface;
-use Magento\Framework\MessageQueue\QueueInterface;
 use Magento\MysqlMq\Model\QueueManagement;
 use Magento\Framework\MessageQueue\EnvelopeFactory;
+use Magento\MysqlMq\Model\ResourceModel\Queue as QueueResourceModel;
 use Psr\Log\LoggerInterface;
 
 /**
  * Queue based on MessageQueue protocol
  */
-class Queue implements QueueInterface
+class Queue implements CountableQueueInterface
 {
     /**
      * @var QueueManagement
@@ -46,6 +48,11 @@ class Queue implements QueueInterface
      */
     private $logger;
 
+    /**
+     * @var QueueResourceModel
+     */
+    private $queueResourceModel;
+
     /**
      * Queue constructor.
      *
@@ -55,21 +62,25 @@ class Queue implements QueueInterface
      * @param string $queueName
      * @param int $interval
      * @param int $maxNumberOfTrials
+     * @param QueueResourceModel|null $queueResourceModel
      */
     public function __construct(
         QueueManagement $queueManagement,
         EnvelopeFactory $envelopeFactory,
         LoggerInterface $logger,
         $queueName,
         $interval = 5,
-        $maxNumberOfTrials = 3
+        $maxNumberOfTrials = 3,
+        ?QueueResourceModel $queueResourceModel = null
     ) {
         $this->queueManagement = $queueManagement;
         $this->envelopeFactory = $envelopeFactory;
         $this->queueName = $queueName;
         $this->interval = $interval;
         $this->maxNumberOfTrials = $maxNumberOfTrials;
         $this->logger = $logger;
+        $this->queueResourceModel = $queueResourceModel
+            ?? ObjectManager::getInstance()->get(QueueResourceModel::class);
     }
 
     /**
@@ -151,4 +162,12 @@ public function push(EnvelopeInterface $envelope)
             [$this->queueName]
         );
     }
+
+    /**
+     * @inheritDoc
+     */
+    public function count(): int
+    {
+        return $this->queueResourceModel->getMessagesCount($this->queueName);
+    }
 }

app/code/Magento/MysqlMq/Model/ResourceModel/Queue.php

@@ -5,6 +5,8 @@
  */
 namespace Magento\MysqlMq\Model\ResourceModel;
 
+use Magento\Framework\DB\Select;
+use Magento\Framework\DB\Sql\Expression;
 use Magento\MysqlMq\Model\QueueManagement;
 
 /**
@@ -240,6 +242,35 @@ public function changeStatus($relationIds, $status)
         );
     }
 
+    /**
+     * Get number of pending messages in the queue
+     *
+     * @param string $queueName
+     * @return int
+     */
+    public function getMessagesCount(string $queueName): int
+    {
+        $connection = $this->getConnection();
+        $select = $connection->select()
+            ->from(
+                ['queue_message' => $this->getMessageTable()],
+            )->join(
+                ['queue_message_status' => $this->getMessageStatusTable()],
+                'queue_message.id = queue_message_status.message_id'
+            )->join(
+                ['queue' => $this->getQueueTable()],
+                'queue.id = queue_message_status.queue_id'
+            )->where(
+                'queue_message_status.status IN (?)',
+                [QueueManagement::MESSAGE_STATUS_NEW, QueueManagement::MESSAGE_STATUS_RETRY_REQUIRED]
+            )->where('queue.name = ?', $queueName);
+
+        $select->reset(Select::COLUMNS);
+        $select->columns(new Expression('COUNT(*)'));
+
+        return (int) $connection->fetchOne($select);
+    }
+
     /**
      * Get name of table storing message statuses and associations to queues.
      *

dev/tests/integration/testsuite/Magento/MysqlMq/Model/Driver/QueueTest.php

@@ -6,6 +6,7 @@
 namespace Magento\MysqlMq\Model\Driver;
 
 use Magento\MysqlMq\Model\Driver\Queue;
+use Magento\MysqlMq\Model\ResourceModel\MessageCollection;
 
 /**
  * Test for MySQL queue driver class.
@@ -43,6 +44,11 @@ protected function tearDown(): void
         /** @var \Magento\Framework\MessageQueue\Config\Data $queueConfig */
         $queueConfig = $this->objectManager->get(\Magento\Framework\MessageQueue\Config\Data::class);
         $queueConfig->reset();
+        $messageCollection = $this->objectManager->create(MessageCollection::class);
+        foreach ($messageCollection as $message) {
+            $message->delete();
+        }
+        parent::tearDown();
     }
 
     /**
@@ -65,4 +71,29 @@ public function testPushAndDequeue()
         $this->assertArrayHasKey('topic_name', $actualMessageProperties);
         $this->assertEquals($topicName, $actualMessageProperties['topic_name']);
     }
+
+    /**
+     * @magentoDataFixture Magento/MysqlMq/_files/queues.php
+     */
+    public function testCount()
+    {
+        /** @var \Magento\Framework\MessageQueue\EnvelopeFactory $envelopFactory */
+        $envelopFactory = $this->objectManager->get(\Magento\Framework\MessageQueue\EnvelopeFactory::class);
+        $messageBody = '{"data": {"body": "Message body"}, "message_id": 1}';
+        $topicName = 'some.topic';
+        $envelop1 = $envelopFactory->create(['body' => $messageBody, 'properties' => ['topic_name' => $topicName]]);
+        $envelop2 = $envelopFactory->create(['body' => $messageBody, 'properties' => ['topic_name' => $topicName]]);
+        $envelop3 = $envelopFactory->create(['body' => $messageBody, 'properties' => ['topic_name' => $topicName]]);
+
+        $this->queue->push($envelop1);
+        $this->queue->push($envelop2);
+        $this->queue->push($envelop3);
+
+        // Take first message in progress and reject
+        $this->queue->reject($this->queue->dequeue());
+        // Take second message in progress
+        $this->queue->dequeue();
+        // Assert that only 2 messages are available in queue (message1 and message3)
+        $this->assertEquals(2, $this->queue->count());
+    }
 }

lib/internal/Magento/Framework/MessageQueue/CountableQueueInterface.php

@@ -0,0 +1,26 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+declare(strict_types=1);
+
+namespace Magento\Framework\MessageQueue;
+
+use Countable;
+
+/**
+ * {@inheritdoc}
+ *
+ * Queue driver that implements this interface must implement count() method
+ * that returns the number of pending messages in the queue
+ */
+interface CountableQueueInterface extends QueueInterface, Countable
+{
+    /**
+     * Get number of pending messages in the queue
+     *
+     * @return int
+     */
+    public function count(): int;
+}