[dbal] Introduce redelivery support based on visibility approach. #581

rosamarsky · 2018-10-23T17:16:59Z

Upd docs

makasim · 2018-10-23T17:40:16Z

pkg/dbal/DbalConsumer.php

    {
+        $this->uuid = $uuid;


makasim · 2018-10-23T17:40:45Z

pkg/dbal/DbalConsumer.php

+        $this->redeliveryDelay = 600; // 10 minutes
+    }
+
+    public function getUuid(): UuidInterface


getConsumerId(): string

makasim · 2018-10-23T17:41:11Z

pkg/dbal/DbalConsumer.php

        $this->context = $context;
        $this->queue = $queue;
        $this->dbal = $this->context->getDbalConnection();

        $this->pollingInterval = 1000;
+        $this->redeliveryDelay = 600; // 10 minutes


should be in milliseconds. 20 minutes looks better to me.

pkg/dbal/DbalConsumer.php

makasim · 2018-10-23T17:44:54Z

pkg/dbal/DbalConsumer.php

    }

    protected function receiveMessage(): ?DbalMessage
    {
+        $this->refreshRedeliveredMessages();


$this->redeliverMessages();

makasim · 2018-10-27T13:23:27Z

pkg/dbal/DbalConsumer.php

     * @var int
     */
-    private $pollingInterval;
+    private $redeliveryDelay = 1200000;


set the value in the constructor.

makasim · 2018-10-27T13:25:44Z

pkg/dbal/DbalConsumer.php

    }

-    protected function receiveMessage(): ?DbalMessage
+    private function receiveMessage(): ?DbalMessage


move the from this method to receiveNoWait. Remove the method after moving

makasim · 2018-10-27T13:36:48Z

pkg/dbal/DbalConsumer.php

@@ -115,33 +97,54 @@ public function reject(Message $message, bool $requeue = false): void

            return;
        }
+
+        $this->deleteMessageByDeliveryId($message->getDeliveryId());


rename to deleteMessage. the method should accept a message instance and throws an exception if delivery id is not set.

makasim · 2018-10-27T13:37:35Z

pkg/dbal/DbalConsumer.php

-            if (false == $dbalMessage) {
+            $now = (int) time();
+            $redeliveryDelay = $this->getRedeliveryDelay() / 1000; // milliseconds to seconds
+            $deliveryId = Uuid::uuid1()->toString();


I'd prefer (string) Uuid::uuid1()

makasim · 2018-10-27T13:39:26Z

pkg/dbal/DbalConsumer.php

+            $deliveryId = Uuid::uuid1()->toString();
+
+            // get top message from the queue
+            $message = $this->fetchMessage($now);


the method could return null, if no message in the queue. but you do not check that a few lines later and assume the message is there.

pkg/dbal/DbalConsumer.php

makasim · 2018-10-27T13:41:50Z

pkg/dbal/DbalConsumer.php


            $this->dbal->commit();

-            if (empty($dbalMessage['time_to_live']) || ($dbalMessage['time_to_live'] / 1000) > microtime(true)) {
+            if (empty($dbalMessage['time_to_live']) || $dbalMessage['time_to_live'] > time()) {


do not expire redelivered messages

pkg/dbal/DbalContext.php

makasim · 2018-10-30T08:07:36Z

pkg/dbal/DbalContext.php


        $table->setPrimaryKey(['id']);
        $table->addIndex(['published_at']);
        $table->addIndex(['queue']);
        $table->addIndex(['priority']);
        $table->addIndex(['delayed_until']);
        $table->addIndex(['priority', 'published_at']);
+        $table->addIndex(['redeliver_after']);
+        $table->addUniqueIndex(['delivery_id']);
+        $table->addIndex(['delivery_id']);


Remove NOT unique index ( there is no need for two indexes)

makasim · 2018-10-30T08:11:40Z

pkg/dbal/DbalConsumer.php

    }

-    protected function receiveMessage(): ?DbalMessage
+    private function deleteMessage(?string $deliveryId): void


Null is not allowed.

I would also add a check that the string is not empty.

makasim · 2018-10-30T08:12:41Z

pkg/dbal/DbalConsumer.php

            ->addOrderBy('priority', 'desc')
+            ->addOrderBy('published_at', 'asc')


could you please test the behavior of sorting when a priority is null

makasim · 2018-10-30T08:17:16Z

@rosamarsky could you update docs? use as an example https://github.com/php-enqueue/enqueue-dev/pull/585/files#diff-1ab2e57b012481304b916b4987c773b5L8

makasim · 2018-10-30T08:47:29Z

The code shared between Consumer and SubscriptionConsumer could go to a trait. Look at how it is done in redis https://github.com/php-enqueue/enqueue-dev/blob/master/pkg/redis/RedisConsumerHelperTrait.php

makasim · 2018-10-30T14:43:52Z

pkg/dbal/DbalConsumer.php

                return null;
            }

-            usleep($this->pollingInterval * 1000);
+            $this->markMessageAsDeliveredToConsumer($message, $deliveryId);


markMessageDelivered

makasim · 2018-10-30T14:45:54Z

pkg/dbal/DbalConsumerHelperTrait.php

+    protected function markMessageAsDeliveredToConsumer(array $message, string $deliveryId): void
+    {
+        $now = time();
+        $redeliveryDelay = $this->getRedeliveryDelay() / 1000; // milliseconds to seconds


the method could not be used in the trait. pass the delay as argument

makasim · 2018-10-30T14:49:35Z

pkg/dbal/DbalConsumer.php

        ;
-
-        $sql = $query->getSQL().' '.$this->dbal->getDatabasePlatform()->getWriteLockSQL();


we need a transaction in both consumer and subscription consumer
both should use getWriteLockSQL

makasim · 2018-10-30T14:51:57Z

pkg/dbal/DbalConsumerHelperTrait.php

+
+    abstract public function getConnection(): Connection;
+
+    protected function fetchMessage(array $queues): ?array


this should be the only one method, it should return an array with a message and queue it was consumed from.

makasim · 2018-10-30T16:56:40Z

pkg/dbal/DbalConsumer.php

+        return $this->context;
+    }
+
+    public function getConnection(): Connection


why public?

makasim · 2018-10-30T16:58:43Z

pkg/dbal/DbalConsumer.php

-        $timeout /= 1000;
-        $startAt = microtime(true);
+        $deliveryId = (string) Uuid::uuid1();
+        $redeliveryDelay = $this->getRedeliveryDelay() / 1000; // milliseconds to seconds


the line states that redeliveryDelay is in seconds.

makasim · 2018-10-30T16:59:59Z

pkg/dbal/DbalConsumer.php

-    public function receiveNoWait(): ?Message
-    {
-        return $this->receiveMessage();
+        return null;
    }

    /**
     * @param DbalMessage $message
     */
    public function acknowledge(Message $message): void
    {


add an exception

InvalidMessageException::assertMessageInstanceOf($message, DbalMessage::class);

makasim · 2018-10-30T17:01:55Z

pkg/dbal/DbalConsumer.php

    }

-    private function fetchMessage(int $now): ?array
+    private function redeliverMessages(): void


the method should be in the consumer trait. It must be called in subscription consumer too.

makasim · 2018-10-30T17:03:06Z

pkg/dbal/DbalConsumerHelperTrait.php

+        try {
+            $now = time();
+
+            $this->getConnection()->beginTransaction();


this should be out of try ... catch block

makasim · 2018-10-30T17:06:43Z

pkg/dbal/DbalConsumerHelperTrait.php

+
+    abstract public function getConnection(): Connection;
+
+    protected function fetchMessage(array $queues, string $deliveryId, int $redeliveryDelay): ?array


What if this method returns an instance of DbalMessage. You have access to context so you can use a $context->convertMessage() method.

makasim · 2018-10-30T17:09:13Z

pkg/dbal/DbalSubscriptionConsumer.php

-        $endAt = time() + $timeout;
+        $now = time();
+        $timeout /= 1000;
+        $deliveryId = (string) Uuid::uuid1();


it should be generated per message, hence the var init should be done before fetchMessage call.

makasim · 2018-10-30T17:09:31Z

pkg/dbal/DbalSubscriptionConsumer.php

+        $now = time();
+        $timeout /= 1000;
+        $deliveryId = (string) Uuid::uuid1();
+        $redeliveryDelay = $this->getRedeliveryDelay() / 1000; // milliseconds to seconds


Already seconds

makasim · 2018-10-30T17:12:00Z

pkg/dbal/DbalSubscriptionConsumer.php

@@ -57,12 +94,8 @@ public function consume(int $timeout = 0): void
                $currentQueueNames = $queueNames;
            }



should call redeliverMessages method. on every cycle.

makasim · 2018-10-31T14:09:21Z

pkg/dbal/DbalConsumer.php

-                return null;
+        // get top message from the queue
+        if ($message = $this->fetchMessage([$this->queue->getQueueName()], $redeliveryDelay)) {
+            if ($message['redelivered'] || empty($message['time_to_live']) || $message['time_to_live'] > time()) {


the expired message has to be deleted from message, I think that expiration logic could be moved to fetchMessage method

I gave it another though, and I now I think we could cook a query that removes expired messages. just like redeliverMessage. the expiration in the code should still be preserved

makasim · 2018-10-31T14:12:22Z

pkg/dbal/DbalConsumer.php

@@ -111,106 +108,34 @@ public function reject(Message $message, bool $requeue = false): void
        InvalidMessageException::assertMessageInstanceOf($message, DbalMessage::class);

        if ($requeue) {
-            $this->context->createProducer()->send($this->queue, $message);
+            $this->getContext()->createProducer()->send($this->queue, $message);


The redelivered flag should be forced false.

makasim · 2018-10-31T14:15:05Z

pkg/dbal/DbalConsumer.php

    }

-    private function fetchMessage(int $now): ?array
+    private function deleteMessage(string $deliveryId): void


the method should be moved to trait.

makasim · 2018-10-31T16:26:50Z

pkg/dbal/DbalConsumer.php

@@ -108,6 +109,8 @@ public function reject(Message $message, bool $requeue = false): void
        InvalidMessageException::assertMessageInstanceOf($message, DbalMessage::class);

        if ($requeue) {
+            $message->setRedelivered(false);


clone message and than force redelivered false

makasim · 2018-10-31T18:53:34Z

Thank you Roman for your work!

makasim changed the title ~~Changed consume process~~ [dbal] Changed consume process Oct 23, 2018

makasim reviewed Oct 23, 2018

View reviewed changes

pkg/dbal/DbalConsumer.php Outdated

{

$this->uuid = $uuid;

Copy link

Member

makasim Oct 23, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

consumerId

makasim reviewed Oct 23, 2018

View reviewed changes

pkg/dbal/DbalConsumer.php Show resolved Hide resolved

makasim reviewed Oct 23, 2018

View reviewed changes

makasim changed the title ~~[dbal] Changed consume process~~ [dbal] Introduce redelivery support based on visibility approach. Oct 23, 2018

rosamarsky force-pushed the dbal-subscription-consumer-improvements branch 4 times, most recently from faa1d8e to 0d44dff Compare October 24, 2018 15:33

makasim reviewed Oct 27, 2018

View reviewed changes

pkg/dbal/DbalConsumer.php Outdated Show resolved Hide resolved

makasim reviewed Oct 27, 2018

View reviewed changes

pkg/dbal/DbalContext.php Outdated Show resolved Hide resolved

rosamarsky added 2 commits October 29, 2018 18:00

Introduce redelivery support based on visibility approach

c641c2a

Consumer minor fixes

5f4ef90

rosamarsky force-pushed the dbal-subscription-consumer-improvements branch from 0d44dff to 5f4ef90 Compare October 29, 2018 16:15

makasim reviewed Oct 30, 2018

View reviewed changes

rosamarsky added 3 commits October 30, 2018 14:25

Updated docs

6532d1e

Moved common logic to trait

10f507d

Fixed message deleting in consumer

697d951

makasim reviewed Oct 30, 2018

View reviewed changes

United methods in trait. Covered by transaction

41823bf

rosamarsky force-pushed the dbal-subscription-consumer-improvements branch from 498ccff to 311bfdf Compare October 30, 2018 16:23

Added write lock

03c40ab

rosamarsky force-pushed the dbal-subscription-consumer-improvements branch from 311bfdf to 03c40ab Compare October 30, 2018 16:26

makasim requested changes Oct 30, 2018

View reviewed changes

rosamarsky force-pushed the dbal-subscription-consumer-improvements branch 2 times, most recently from 2a1d940 to 84dca99 Compare October 30, 2018 18:05

Fixes.

50f3f07

rosamarsky force-pushed the dbal-subscription-consumer-improvements branch from 84dca99 to 50f3f07 Compare October 30, 2018 18:13

makasim reviewed Oct 31, 2018

View reviewed changes

rosamarsky force-pushed the dbal-subscription-consumer-improvements branch from e979200 to a82d298 Compare October 31, 2018 16:12

makasim reviewed Oct 31, 2018

View reviewed changes

rosamarsky force-pushed the dbal-subscription-consumer-improvements branch 3 times, most recently from 189ed8c to 24129f1 Compare October 31, 2018 17:55

Deleting of expired messages

239161b

rosamarsky force-pushed the dbal-subscription-consumer-improvements branch from 24129f1 to 239161b Compare October 31, 2018 17:59

makasim merged commit dee836c into php-enqueue:master Oct 31, 2018

makasim added this to the 0.9 milestone Nov 15, 2018

		->addOrderBy('priority', 'desc')
		->addOrderBy('published_at', 'asc')

		;

		$sql = $query->getSQL().' '.$this->dbal->getDatabasePlatform()->getWriteLockSQL();


		abstract public function getConnection(): Connection;

		protected function fetchMessage(array $queues): ?array

		@@ -57,12 +94,8 @@ public function consume(int $timeout = 0): void
		$currentQueueNames = $queueNames;
		}

[dbal] Introduce redelivery support based on visibility approach. #581

[dbal] Introduce redelivery support based on visibility approach. #581

Conversation

rosamarsky commented Oct 23, 2018 • edited by makasim Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

makasim Oct 27, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

makasim commented Oct 30, 2018

makasim commented Oct 30, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

makasim commented Oct 31, 2018

rosamarsky commented Oct 23, 2018 •

edited by makasim

Loading

makasim Oct 27, 2018 •

edited

Loading