make delivery_id field binary, review indexes.

Author: makasim

pkg/dbal/DbalConsumer.php

@@ -11,6 +11,7 @@
 use Interop\Queue\Impl\ConsumerPollingTrait;
 use Interop\Queue\Message;
 use Interop\Queue\Queue;
+use Ramsey\Uuid\Uuid;
 
 class DbalConsumer implements Consumer
 {
@@ -138,8 +139,8 @@ private function deleteMessage(string $deliveryId): void
 
         $this->getConnection()->delete(
             $this->getContext()->getTableName(),
-            ['delivery_id' => $deliveryId],
-            ['delivery_id' => Type::STRING]
+            ['delivery_id' => Uuid::fromString($deliveryId)->getBytes()],
+            ['delivery_id' => Type::BINARY]
         );
     }
 }

pkg/dbal/DbalConsumerHelperTrait.php

@@ -22,16 +22,16 @@ abstract protected function getConnection(): Connection;
     protected function fetchMessage(array $queues, int $redeliveryDelay): ?array
     {
         $now = time();
-        $deliveryId = (string) Uuid::uuid1();
+        $deliveryId = Uuid::uuid1();
 
         $endAt = microtime(true) + 0.2; // add 200ms
 
         $select = $this->getConnection()->createQueryBuilder()
             ->select('id')
             ->from($this->getContext()->getTableName())
-            ->andWhere('delivery_id IS NULL')
-            ->andWhere('delayed_until IS NULL OR delayed_until <= :delayedUntil')
             ->andWhere('queue IN (:queues)')
+            ->andWhere('delayed_until IS NULL OR delayed_until <= :delayedUntil')
+            ->andWhere('delivery_id IS NULL')
             ->addOrderBy('priority', 'asc')
             ->addOrderBy('published_at', 'asc')
             ->setParameter('delayedUntil', $now, ParameterType::INTEGER)
@@ -44,11 +44,11 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?array
             ->set('redeliver_after', ':redeliverAfter')
             ->andWhere('id = :messageId')
             ->andWhere('delivery_id IS NULL')
-            ->setParameter('deliveryId', $deliveryId, Type::STRING)
+            ->setParameter('deliveryId', $deliveryId->getBytes(), Type::BINARY)
             ->setParameter('redeliverAfter', $now + $redeliveryDelay, Type::BIGINT)
         ;
 
-        while (microtime(true) < $endAt) {
+        while (microtime(true) >= $endAt) {
             $result = $select->execute()->fetch();
             if (empty($result)) {
                 return null;
@@ -63,7 +63,7 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?array
                     ->select('*')
                     ->from($this->getContext()->getTableName())
                     ->andWhere('delivery_id = :deliveryId')
-                    ->setParameter('deliveryId', $deliveryId, Type::STRING)
+                    ->setParameter('deliveryId', $deliveryId->getBytes(), Type::BINARY)
                     ->setMaxResults(1)
                     ->execute()
                     ->fetch()
@@ -88,18 +88,19 @@ protected function redeliverMessages(): void
             return;
         }
 
-        $this->getConnection()->createQueryBuilder()
+        $update = $this->getConnection()->createQueryBuilder()
             ->update($this->getContext()->getTableName())
             ->set('delivery_id', ':deliveryId')
             ->set('redelivered', ':redelivered')
-            ->andWhere('delivery_id IS NOT NULL')
             ->andWhere('redeliver_after < :now')
-            ->setParameter(':now', (int) time(), Type::BIGINT)
-            ->setParameter('deliveryId', null, Type::STRING)
+            ->andWhere('delivery_id IS NOT NULL')
+            ->setParameter(':now', time(), Type::BIGINT)
+            ->setParameter('deliveryId', null, Type::BINARY)
             ->setParameter('redelivered', true, Type::BOOLEAN)
-            ->execute()
         ;
 
+        $update->execute();
+
         $this->redeliverMessagesLastExecutedAt = microtime(true);
     }
 
@@ -111,14 +112,17 @@ protected function removeExpiredMessages(): void
             return;
         }
 
-        $this->getConnection()->createQueryBuilder()
+        $update = $this->getConnection()->createQueryBuilder()
             ->delete($this->getContext()->getTableName())
             ->andWhere('(time_to_live IS NOT NULL) AND (time_to_live < :now)')
-            ->andWhere('(redelivered = false OR delivery_id IS NULL)')
+            ->andWhere('delivery_id IS NULL')
+            ->andWhere('redelivered = false')
+
             ->setParameter(':now', (int) time(), Type::BIGINT)
-            ->execute()
         ;
 
+        $update->execute();
+
         $this->removeExpiredMessagesLastExecutedAt = microtime(true);
     }
 }

pkg/dbal/DbalContext.php

@@ -17,6 +17,7 @@
 use Interop\Queue\Queue;
 use Interop\Queue\SubscriptionConsumer;
 use Interop\Queue\Topic;
+use Ramsey\Uuid\Uuid;
 
 class DbalContext implements Context
 {
@@ -160,7 +161,7 @@ public function convertMessage(array $dbalMessage): DbalMessage
             $dbalMessageObj->setPublishedAt((int) $dbalMessage['published_at']);
         }
         if (isset($dbalMessage['delivery_id'])) {
-            $dbalMessageObj->setDeliveryId((string) $dbalMessage['delivery_id']);
+            $dbalMessageObj->setDeliveryId(Uuid::fromBytes($dbalMessage['delivery_id'])->toString());
         }
         if (isset($dbalMessage['redeliver_after'])) {
             $dbalMessageObj->setRedeliverAfter((int) $dbalMessage['redeliver_after']);
@@ -229,20 +230,13 @@ public function createDataBaseTable(): void
         $table->addColumn('priority', Type::SMALLINT, ['notnull' => false]);
         $table->addColumn('delayed_until', Type::BIGINT, ['notnull' => false]);
         $table->addColumn('time_to_live', Type::BIGINT, ['notnull' => false]);
-        $table->addColumn('delivery_id', Type::STRING, ['notnull' => false]);
+        $table->addColumn('delivery_id', Type::BINARY, ['length' => 16, 'fixed' => true, 'notnull' => false]);
         $table->addColumn('redeliver_after', Type::BIGINT, ['notnull' => false]);
 
         $table->setPrimaryKey(['id']);
-        $table->addUniqueIndex(['delivery_id']);
+        $table->addIndex(['priority', 'published_at', 'queue', 'delivery_id', 'delayed_until', 'id']);
 
-        // try to select a message index
-        $table->addIndex(['delivery_id, delayed_until, queue']);
-        $table->addIndex(['priority', 'published_at']);
-
-        // redeliver failed messages
-        $table->addIndex(['delivery_id', 'redeliver_after']);
-
-        // remove expired messages
+        $table->addIndex(['redeliver_after', 'delivery_id']);
         $table->addIndex(['time_to_live', 'delivery_id']);
 
         $sm->createTable($table);

pkg/dbal/DbalSubscriptionConsumer.php

@@ -75,6 +75,7 @@ public function consume(int $timeout = 0): void
         }
 
         $timeout /= 1000;
+        $now = time();
         $redeliveryDelay = $this->getRedeliveryDelay() / 1000; // milliseconds to seconds
 
         $currentQueueNames = [];
@@ -83,7 +84,6 @@ public function consume(int $timeout = 0): void
                 $currentQueueNames = $queueNames;
             }
 
-            $now = time();
             $this->removeExpiredMessages();
             $this->redeliverMessages();