Implement Retry into Queue Lib

composer.json

@@ -15,7 +15,7 @@
         "psr-4": {"Utopia\\Queue\\": "src/Queue"}
     },
     "autoload-dev": {
-        "psr-4": {"Utopia\\Tests\\": "tests/Database"}
+        "psr-4": {"Tests\\E2E\\": "tests/Queue/e2e"}
     },
     "scripts":{
         "test": "phpunit",

phpunit.xml

@@ -10,7 +10,7 @@
         >
     <testsuites>
         <testsuite name="E2E">
-            <directory>./tests/Queue/e2e/</directory>
+            <directory>./tests/Queue/e2e/Adapter</directory>
         </testsuite>
     </testsuites>
 </phpunit>

src/Queue/Client.php

@@ -26,14 +26,47 @@ public function enqueue(array $payload): bool
         return $this->connection->leftPushArray("{$this->namespace}.queue.{$this->queue}", $payload);
     }
 
+    public function retry(int $limit = null): void
+    {
+        $start = \time();
+        $processed = 0;
+
+        while (true) {
+            $pid = $this->connection->rightPop("{$this->namespace}.failed.{$this->queue}", 5);
+
+            if ($pid === false) {
+                break;
+            }
+
+            $job = $this->getJob($pid);
+
+            if ($job === false) {
+                break;
+            }
+
+            if ($job->getTimestamp() >= $start) {
+                break;
+            }
+
+            if ($limit !== null && $processed >= $limit) {
+                break;
+            }
+
+            $this->enqueue($job->getPayload());
+            $processed++;
+        }
+    }
+
     public function getJob(string $pid): Message|false
     {
-        $job = $this->connection->get("{$this->namespace}.jobs.{$this->queue}.{$pid}");
+        $value = $this->connection->get("{$this->namespace}.jobs.{$this->queue}.{$pid}");
 
-        if ($job === false) {
+        if ($value === false) {
             return false;
         }
 
+        $job = json_decode($value, true);
+
         return new Message($job);
     }
 

src/Queue/Connection/Redis.php

@@ -3,7 +3,6 @@
 namespace Utopia\Queue\Connection;
 
 use Utopia\Queue\Connection;
-use Utopia\Queue\Message;
 
 class Redis implements Connection
 {
@@ -69,7 +68,7 @@ public function rightPopArray(string $queue, int $timeout): array|false
             return false;
         }
 
-        return json_decode($response, true);
+        return json_decode($response, true) ?? false;
     }
 
     public function rightPop(string $queue, int $timeout): string|false
@@ -91,7 +90,7 @@ public function leftPopArray(string $queue, int $timeout): array|false
             return false;
         }
 
-        return json_decode($response[1], true);
+        return json_decode($response[1], true) ?? false;
     }
 
     public function leftPop(string $queue, int $timeout): string|false
@@ -152,11 +151,11 @@ public function decrement(string $key): int
 
     public function listRange(string $key, int $total, int $offset): array
     {
-        $start = $offset - 1;
-        $end = ($total + $offset) -1;
-        $results = $this->getRedis()->lrange($key, $start, $end);
+        $start = $offset;
+        $end = $start + $total - 1;
+        $results = $this->getRedis()->lRange($key, $start, $end);
 
-        return array_map(fn (array $job) => new Message($job), $results);
+        return $results;
     }
 
     public function ping(): bool

tests/Queue/e2e/AdapterTest.php → tests/Queue/e2e/Adapter/Base.php

@@ -1,17 +1,13 @@
 <?php
 
-namespace Utopia\Tests;
+namespace Tests\E2E\Adapter;
 
 use PHPUnit\Framework\TestCase;
 use Utopia\Queue\Client;
-use Utopia\Queue\Connection\Redis;
 
-use function Swoole\Coroutine\go;
-use function Swoole\Coroutine\run;
-
-class SwooleTest extends TestCase
+abstract class Base extends TestCase
 {
-    private array $payloads;
+    protected array $payloads;
 
     public function setUp(): void
     {
@@ -58,16 +54,16 @@ public function setUp(): void
         ];
     }
 
+    /**
+     * @return Client
+     */
+    abstract protected function getClient(): Client;
+
     public function testEvents(): void
     {
-        $connection = new Redis('redis', 6379);
-
-        $this->assertTrue($connection->ping());
-
-        $client = new Client('workerman', $connection);
+        $client = $this->getClient();
         $client->resetStats();
 
-
         foreach ($this->payloads as $payload) {
             $this->assertTrue($client->enqueue($payload));
         }
@@ -81,26 +77,49 @@ public function testEvents(): void
         $this->assertEquals(7, $client->sumSuccessfulJobs());
     }
 
-    public function testSwoole(): void
+    /**
+     * @depends testEvents
+     */
+    public function testRetry(): void
     {
-        $connection = new Redis('redis', 6379);
+        $client = $this->getClient();
+        $client->resetStats();
+
+        $client->enqueue([
+            'type' => 'test_exception',
+            'id' => 1
+        ]);
+        $client->enqueue([
+            'type' => 'test_exception',
+            'id' => 2
+        ]);
+        $client->enqueue([
+            'type' => 'test_exception',
+            'id' => 3
+        ]);
+        $client->enqueue([
+            'type' => 'test_exception',
+            'id' => 4
+        ]);
 
-        run(function () use ($connection) {
-            $client = new Client('swoole', $connection);
-            go(function () use ($client) {
-                $client->resetStats();
+        sleep(1);
+
+        $this->assertEquals(4, $client->sumTotalJobs());
+        $this->assertEquals(0, $client->sumProcessingJobs());
+        $this->assertEquals(4, $client->sumFailedJobs());
+        $this->assertEquals(0, $client->sumSuccessfulJobs());
 
-                foreach ($this->payloads as $payload) {
-                    $this->assertTrue($client->enqueue($payload));
-                }
+        $client->resetStats();
 
-                sleep(1);
+        $client->retry();
 
-                $this->assertEquals(8, $client->sumTotalJobs());
-                $this->assertEquals(0, $client->sumProcessingJobs());
-                $this->assertEquals(1, $client->sumFailedJobs());
-                $this->assertEquals(7, $client->sumSuccessfulJobs());
-            });
-        });
+        sleep(1);
+
+        // Retry will retry ALL failed jobs regardless of if they are still tracked in stats
+        // Meaning this test has 5 failed jobs due to the previous tests.
+        $this->assertEquals(5, $client->sumTotalJobs());
+        $this->assertEquals(0, $client->sumProcessingJobs());
+        $this->assertEquals(5, $client->sumFailedJobs());
+        $this->assertEquals(0, $client->sumSuccessfulJobs());
     }
 }

tests/Queue/e2e/Adapter/SwooleTest.php

@@ -0,0 +1,45 @@
+<?php
+
+namespace Tests\E2E\Adapter;
+
+use Utopia\Queue\Client;
+use Utopia\Queue\Connection\Redis;
+
+use function Co\run;
+
+class SwooleTest extends Base
+{
+    protected function getClient(): Client
+    {
+        $connection = new Redis('redis', 6379);
+        $client = new Client('swoole', $connection);
+
+        return $client;
+    }
+
+    /**
+     * @depends testRetry
+     */
+    protected function testSwooleConcurrency(): void
+    {
+        $connection = new Redis('redis', 6379);
+
+        run(function () use ($connection) {
+            $client = new Client('swoole', $connection);
+            go(function () use ($client) {
+                $client->resetStats();
+
+                foreach ($this->payloads as $payload) {
+                    $this->assertTrue($client->enqueue($payload));
+                }
+
+                sleep(1);
+
+                $this->assertEquals(8, $client->sumTotalJobs());
+                $this->assertEquals(0, $client->sumProcessingJobs());
+                $this->assertEquals(1, $client->sumFailedJobs());
+                $this->assertEquals(7, $client->sumSuccessfulJobs());
+            });
+        });
+    }
+}

tests/Queue/e2e/Adapter/WorkermanTest.php

@@ -0,0 +1,17 @@
+<?php
+
+namespace Tests\E2E\Adapter;
+
+use Utopia\Queue\Client;
+use Utopia\Queue\Connection\Redis;
+
+class WorkermanTest extends Base
+{
+    protected function getClient(): Client
+    {
+        $connection = new Redis('redis', 6379);
+        $client = new Client('workerman', $connection);
+
+        return $client;
+    }
+}