Add $retryTime property for DbQueue and AwsSqsQueue

Author: ElGigi

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [1.0.0-beta7] - 2025-01-27
 
+### Added
+
+- `$retryTime` property for `DbQueue` and `AwsSqsQueue`
+
 ### Changed
 
 - Methods `QueueInterface::push()` and `QueueInterface::pushRaw()`, accept `\DateTimeInterface` or `\DateInterval` for `$delay` parameter

README.md

@@ -161,6 +161,7 @@ $queue = new DbQueue(
     connection: $dbConnection, // Database connection
     name: 'default',           // Queue name
     tableName: 'queue_jobs',   // Name of MySQL table
+    retryTime: 30,             // Time to wait after failed job
     maxAttempts: 5,            // Maximum attempts of a job
 );
 ```
@@ -204,7 +205,7 @@ CREATE
   TRIGGER `queue_jobs_AFTER_DELETE` AFTER DELETE ON `queue_jobs`
   FOR EACH ROW BEGIN
     INSERT INTO `queue_jobs_done` (`job_id`, `create_time`, `queue`, `availability_time`, `attempts`, `lock_time`, `payload`)
-    VALUES (OLD.`job_id`, OLD.`create_time`, OLD.`queue`, OLD.`availability_time`, OLD.`attempts`, OLD.`lock_time`, OLD.`payload`);
+    VALUES (OLD.`job_id`, OLD.`create_time`, OLD.`queue`, OLD.`availability_time`, OLD.`attempts`, IFNULL(OLD.`lock_time`, CURRENT_TIMESTAMP), OLD.`payload`);
   END;$$
 DELIMITER ;
 ```
@@ -258,5 +259,6 @@ $queue = new AwsSqsQueue(
     sqsClient: new SqsClient(...), // Database connection
     name: 'default',               // Queue name
     queueUrl: '...',               // AWS queue URL
+    retryTime: 30,                 // Time to wait after failed job
 );
 ```

src/Queue/AbstractQueue.php

@@ -85,12 +85,6 @@ protected function getAvailableDateTime(DateTimeInterface|DateInterval|int $dela
      */
     protected function getDelayInSeconds(DateTimeInterface $dateTime): int
     {
-        $diff = $this->now()->diff($dateTime);
-
-        return
-            $diff->days * (24 * 60 * 60) +
-            $diff->h * (60 * 60) +
-            $diff->i * (60) +
-            $diff->s;
+        return $dateTime->getTimestamp() - $this->now()->getTimestamp();
     }
 }

src/Queue/AwsSqsQueue.php

@@ -30,6 +30,7 @@ public function __construct(
         private SqsClient $sqsClient,
         private string $queueUrl,
         string $name = 'default',
+        private int $retryTime = 30,
     ) {
         parent::__construct($name);
     }
@@ -88,6 +89,7 @@ public function consume(): ?SqsJob
     {
         $result = $this->sqsClient->receiveMessage([
             'QueueUrl' => $this->queueUrl,
+            'VisibilityTimeout' => $this->retryTime,
             'AttributeNames' => ['ApproximateReceiveCount'],
         ])->toArray()['Messages'][0] ?? null;
 
@@ -117,7 +119,7 @@ public function push(JobDescriptorInterface $jobDescriptor, DateTimeInterface|Da
     public function pushRaw(mixed $payload, DateTimeInterface|DateInterval|int $delay = 0): string
     {
         $result = $this->sqsClient->sendMessage([
-            'DelaySeconds' => $delay,
+            'DelaySeconds' => $this->getDelayInSeconds($this->getAvailableDateTime($delay)),
             'MessageBody' => json_encode($payload),
             'QueueUrl' => $this->queueUrl,
         ]);

src/Queue/DbQueue.php

@@ -25,6 +25,7 @@
 use Hector\Connection\Connection;
 use Hector\Query\Component\Order;
 use Hector\Query\QueryBuilder;
+use Hector\Query\Statement\Conditions;
 
 class_exists(QueryBuilder::class) || throw QueueManagerException::missingPackage('hectororm/query');
 
@@ -34,11 +35,17 @@ public function __construct(
         private Connection $connection,
         string $name = 'default',
         private string $tableName = 'queue_jobs',
+        private int $retryTime = 30,
         private int $maxAttempts = 5,
     ) {
         parent::__construct($name);
     }
 
+    private function getRetryDateTimeLimit(): DateTimeInterface
+    {
+        return $this->now()->sub(new DateInterval('PT' . $this->retryTime . 'S'));
+    }
+
     private function getQueryBuilder(): QueryBuilder
     {
         $builder = new QueryBuilder($this->connection);
@@ -48,16 +55,26 @@ private function getQueryBuilder(): QueryBuilder
         return $builder;
     }
 
+    private function addBuilderConditions(QueryBuilder $builder): QueryBuilder
+    {
+        $retryLimit = $this->getRetryDateTimeLimit();
+
+        return $builder
+            ->whereLessThanOrEqual('availability_time', $this->now()->format('Y-m-d H:i:s'))
+            ->whereLessThan('attempts', $this->maxAttempts)
+            ->where(function (Conditions $where) use ($retryLimit): void {
+                $where
+                    ->whereNull('lock_time')
+                    ->orWhere('lock_time', '<', $retryLimit->format('Y-m-d H:i:s'));
+            });
+    }
+
     /**
      * @inheritDoc
      */
     public function size(): int
     {
-        return $this->getQueryBuilder()
-            ->whereLessThanOrEqual('availability_time', $this->now()->format('Y-m-d H:i:s'))
-            ->whereNull('lock_time')
-            ->whereLessThan('attempts', $this->maxAttempts)
-            ->count();
+        return $this->addBuilderConditions($this->getQueryBuilder())->count();
     }
 
     /**
@@ -68,10 +85,7 @@ public function consume(): ?DbJob
         $attempts = 0;
 
         do {
-            $jobRaw = $this->getQueryBuilder()
-                ->whereLessThanOrEqual('availability_time', $this->now()->format('Y-m-d H:i:s'))
-                ->whereNull('lock_time')
-                ->whereLessThan('attempts', $this->maxAttempts)
+            $jobRaw = $this->addBuilderConditions($this->getQueryBuilder())
                 ->orderBy('job_id', Order::ORDER_ASC)
                 ->limit(1)
                 ->fetchOne();

tests/Queue/AwsSqsQueueTest.php

@@ -75,6 +75,7 @@ public function testConsumeReturnsJob()
                 [
                     [
                         'QueueUrl' => 'https://sqs.us-east-1.amazonaws.com/123456789012/testQueue',
+                        'VisibilityTimeout' => 30,
                         'AttributeNames' => ['ApproximateReceiveCount'],
                     ]
                 ]
@@ -95,6 +96,7 @@ public function testConsumeThrowsChecksumException()
                 [
                     [
                         'QueueUrl' => 'https://sqs.us-east-1.amazonaws.com/123456789012/testQueue',
+                        'VisibilityTimeout' => 30,
                         'AttributeNames' => ['ApproximateReceiveCount'],
                     ]
                 ]